diff --git a/api/.env.template b/api/.env.template index 5a4c77a4ee..2707e9cfc5 100644 --- a/api/.env.template +++ b/api/.env.template @@ -58,6 +58,8 @@ LISTING_OPEN_DATE_NOTIFICATION_CRON_STRING=0 9 * * * LOTTERY_PUBLISH_PROCESSING_CRON_STRING=58 23 * * * # controls the repetition of the lottery cron job LOTTERY_PROCESSING_CRON_STRING=0 * * * * +# controls the repetition of the stuck background jobs recovery job +BACKGROUND_JOBS_RECOVERY_CRON_STRING=0/30 * * * * # how many days till lottery data expires LOTTERY_DAYS_TILL_EXPIRY=45 # controls the repetition of the msq retire cron job (should occur after LISTING_PROCESSING_CRON_STRING) diff --git a/api/src/modules/background-jobs.module.ts b/api/src/modules/background-jobs.module.ts index fac5f2aba1..bcadedb2e3 100644 --- a/api/src/modules/background-jobs.module.ts +++ b/api/src/modules/background-jobs.module.ts @@ -1,13 +1,14 @@ -import { Module } from '@nestjs/common'; +import { Logger, Module } from '@nestjs/common'; import { PrismaModule } from './prisma.module'; import { S3Module } from './s3.module'; import { PermissionModule } from './permission.module'; import { BackgroundJobsController } from '../controllers/background-jobs.controller'; import { BackgroundJobsService } from '../services/background-jobs.service'; +import { CronJobModule } from './cron-job.module'; @Module({ - imports: [PrismaModule, S3Module, PermissionModule], - providers: [BackgroundJobsService], + imports: [PrismaModule, S3Module, PermissionModule, CronJobModule], + providers: [BackgroundJobsService, Logger], controllers: [BackgroundJobsController], exports: [BackgroundJobsService], }) diff --git a/api/src/services/background-jobs.service.ts b/api/src/services/background-jobs.service.ts index ef644a2026..68f45e4c7f 100644 --- a/api/src/services/background-jobs.service.ts +++ b/api/src/services/background-jobs.service.ts @@ -1,6 +1,8 @@ import { ConflictException, + Inject, Injectable, + Logger, NotFoundException, } from '@nestjs/common'; import { PrismaService } from './prisma.service'; @@ -13,15 +15,29 @@ import { BackgroundJobCreate } from '../dtos/background-jobs/background-job-crea import { PermissionService } from './permission.service'; import { permissionActions } from '../enums/permissions/permission-actions-enum'; import { SuccessDTO } from '../dtos/shared/success.dto'; +import { CronJobService } from './cron-job.service'; +const BACKGROUND_JOBS_RECOVERY_JOB_NAME = 'BACKGROUND_JOBS_RECOVERY_CRON_JOB'; +export const BACKGROUND_JOB_STALE_TIME_IN_MINUTES = 30; @Injectable() export class BackgroundJobsService { constructor( private readonly prismaService: PrismaService, private readonly s3Service: S3Service, + private readonly cronJobService: CronJobService, private readonly permissionService: PermissionService, + @Inject(Logger) + private logger = new Logger(BackgroundJobsService.name), ) {} + onModuleInit() { + this.cronJobService.startCronJob( + BACKGROUND_JOBS_RECOVERY_JOB_NAME, + process.env.BACKGROUND_JOBS_RECOVERY_CRON_STRING, + this.recoverStuckJobCronJob.bind(this), + ); + } + /** * Creates an instance of a background job runner for a listing * @param dto - background job creation DTO @@ -144,4 +160,61 @@ export class BackgroundJobsService { success: !!activeJob, }; } + + async recoverStuckJobCronJob(): Promise { + const logName = 'recoverStuckJobCron'; + this.logger.warn(`${logName} job running`); + const currentTime = new Date(); + + this.cronJobService.markCronJobAsStarted(BACKGROUND_JOBS_RECOVERY_JOB_NAME); + + const runningJobs = await this.prismaService.backgroundJob.findMany({ + select: { + id: true, + updatedAt: true, + }, + where: { + status: BackgroundJobStatusEnum.processing, + }, + }); + + if (!runningJobs.length) { + this.logger.warn('No jobs are currently running'); + return { + success: true, + }; + } + + const markedForDeletion = []; + runningJobs.forEach((job) => { + if ( + (currentTime.getTime() - job.updatedAt.getTime()) / 60000 > + BACKGROUND_JOB_STALE_TIME_IN_MINUTES + ) { + markedForDeletion.push(job.id); + } + }); + + this.logger.warn( + `${markedForDeletion.length} running jobs have been marked for deletion`, + ); + + const deletedJobs = await this.prismaService.backgroundJob.deleteMany({ + where: { + id: { + in: markedForDeletion, + }, + }, + }); + + if (deletedJobs.count !== markedForDeletion.length) { + this.logger.error( + `Failed to delete all the marked jobs (${deletedJobs.count}/${markedForDeletion.length} jobs deleted)`, + ); + } + + return { + success: true, + }; + } } diff --git a/api/test/unit/services/background-jobs.service.spec.ts b/api/test/unit/services/background-jobs.service.spec.ts index c4b09f36a0..3bf793ef13 100644 --- a/api/test/unit/services/background-jobs.service.spec.ts +++ b/api/test/unit/services/background-jobs.service.spec.ts @@ -2,16 +2,22 @@ import { Test, TestingModule } from '@nestjs/testing'; import { ConflictException, ForbiddenException, + Logger, NotFoundException, } from '@nestjs/common'; +import { SchedulerRegistry } from '@nestjs/schedule'; import { BackgroundJobStatusEnum } from '@prisma/client'; import { randomUUID } from 'crypto'; -import { BackgroundJobsService } from '../../../src/services/background-jobs.service'; +import { + BACKGROUND_JOB_STALE_TIME_IN_MINUTES, + BackgroundJobsService, +} from '../../../src/services/background-jobs.service'; import { PrismaService } from '../../../src/services/prisma.service'; import { S3Service } from '../../../src/services/s3.service'; import { PermissionService } from '../../../src/services/permission.service'; import { BackgroundJob } from '../../../src/dtos/background-jobs/background-job.dto'; import { UserRoleEnum } from '../../../src/enums/permissions/user-role-enum'; +import { CronJobService } from '../../../src/services/cron-job.service'; const listingId = randomUUID(); const jobId = randomUUID(); @@ -48,6 +54,9 @@ describe('Background Jobs Service Tests', () => { providers: [ BackgroundJobsService, PrismaService, + Logger, + SchedulerRegistry, + CronJobService, { provide: S3Service, useValue: {} }, { provide: PermissionService, @@ -196,4 +205,53 @@ describe('Background Jobs Service Tests', () => { }); }); }); + + describe('recoverStuckJobCronJob', () => { + it(`should delete only jobs running fro more than ${BACKGROUND_JOB_STALE_TIME_IN_MINUTES} minutes`, async () => { + const runningValidJobId = randomUUID(); + const runningStaleJobId = randomUUID(); + + prisma.cronJob.findFirst = jest + .fn() + .mockResolvedValue({ id: randomUUID() }); + prisma.cronJob.update = jest.fn().mockResolvedValue(true); + prisma.backgroundJob.findMany = jest.fn().mockReturnValue([ + { + id: runningValidJobId, + updatedAt: new Date(), + }, + { + id: runningStaleJobId, + updatedAt: new Date( + new Date().getTime() - + (BACKGROUND_JOB_STALE_TIME_IN_MINUTES + 1) * 60000, + ), + }, + ]); + prisma.backgroundJob.deleteMany = jest.fn().mockResolvedValue({ + count: 1, + }); + + await service.recoverStuckJobCronJob(); + + expect(prisma.backgroundJob.findMany).toHaveBeenCalledTimes(1); + expect(prisma.backgroundJob.findMany).toHaveBeenCalledWith({ + select: { + id: true, + updatedAt: true, + }, + where: { + status: BackgroundJobStatusEnum.processing, + }, + }); + expect(prisma.backgroundJob.deleteMany).toHaveBeenCalledTimes(1); + expect(prisma.backgroundJob.deleteMany).toHaveBeenCalledWith({ + where: { + id: { + in: [runningStaleJobId], + }, + }, + }); + }); + }); });