diff --git a/src/backend/src/app.module.ts b/src/backend/src/app.module.ts index 8b352cf..12caa6e 100644 --- a/src/backend/src/app.module.ts +++ b/src/backend/src/app.module.ts @@ -55,7 +55,20 @@ import { ThrottlerGuard, ThrottlerModule } from '@nestjs/throttler'; imports: [ConfigModule], useFactory: async (configService: ConfigService) => ({ defaultJobOptions: { - removeOnComplete: true, + attempts: 2, + backoff: { + type: 'exponential', + delay: 30_000, + }, + removeOnComplete: { + age: 3_600, + count: 1_000, + }, + removeOnFail: { + age: 86_400, + count: 1_000, + }, + timeout: 10 * 60_000, }, connection: { host: configService.get(EnvironmentEnum.REDIS_HOST), diff --git a/src/backend/src/track/track-download.processor.ts b/src/backend/src/track/track-download.processor.ts index de1b7d9..3f88d56 100644 --- a/src/backend/src/track/track-download.processor.ts +++ b/src/backend/src/track/track-download.processor.ts @@ -1,4 +1,5 @@ import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Logger } from '@nestjs/common'; import { Job } from 'bullmq'; import { TrackService } from './track.service'; import { TrackEntity } from './track.entity'; @@ -13,15 +14,29 @@ function getDownloadsPerMinute(): number { return Math.max(1, Math.min(30, Math.floor(parsed))); } -@Processor('track-download-processor') +@Processor('track-download-processor', { + concurrency: 1, +}) export class TrackDownloadProcessor extends WorkerHost { + private readonly logger = new Logger(TrackDownloadProcessor.name); + constructor(private readonly trackService: TrackService) { super(); } async process(job: Job): Promise { + if (!job.data?.id) { + this.logger.warn(`Skipping malformed download job ${job.id}`); + return; + } + const maxPerMinute = getDownloadsPerMinute(); const sleepMs = Math.floor(60000 / maxPerMinute); + + this.logger.debug( + `Processing download job ${job.id} for track ${job.data.id}; pacing ${sleepMs}ms`, + ); + await new Promise((res) => setTimeout(res, sleepMs)); await this.trackService.downloadFromYoutube(job.data); } diff --git a/src/backend/src/track/track-search.processor.ts b/src/backend/src/track/track-search.processor.ts index 0f16a26..0336193 100644 --- a/src/backend/src/track/track-search.processor.ts +++ b/src/backend/src/track/track-search.processor.ts @@ -1,15 +1,26 @@ import { Processor, WorkerHost } from '@nestjs/bullmq'; +import { Logger } from '@nestjs/common'; import { Job } from 'bullmq'; import { TrackService } from './track.service'; import { TrackEntity } from './track.entity'; -@Processor('track-search-processor') +@Processor('track-search-processor', { + concurrency: 2, +}) export class TrackSearchProcessor extends WorkerHost { + private readonly logger = new Logger(TrackSearchProcessor.name); + constructor(private readonly trackService: TrackService) { super(); } async process(job: Job): Promise { + if (!job.data?.id) { + this.logger.warn(`Skipping malformed search job ${job.id}`); + return; + } + + this.logger.debug(`Processing search job ${job.id} for track ${job.data.id}`); await this.trackService.findOnYoutube(job.data); } } diff --git a/src/backend/src/track/track.service.ts b/src/backend/src/track/track.service.ts index 6a43006..8780989 100644 --- a/src/backend/src/track/track.service.ts +++ b/src/backend/src/track/track.service.ts @@ -58,9 +58,7 @@ export class TrackService { async create(track: TrackEntity, playlist?: PlaylistEntity): Promise { const savedTrack = await this.repository.save({ ...track, playlist }); - await this.trackSearchQueue.add('', savedTrack, { - jobId: `id-${savedTrack.id}`, - }); + await this.enqueueSearch(savedTrack.id); this.io.emit(WsTrackOperation.New, { track: savedTrack, playlistId: playlist.id, @@ -74,23 +72,30 @@ export class TrackService { async retry(id: number): Promise { const track = await this.get(id); - const existingJob = await this.trackSearchQueue.getJob(`id-${id}`); - if (existingJob) { - this.logger.warn(`Search job already exists for track ${id}`); + if (!track) { + this.logger.warn(`Cannot retry missing track ${id}`); return; } - await this.trackSearchQueue.add('', track, { - jobId: `id-${id}`, + await this.enqueueSearch(id); + await this.update(id, { + ...track, + error: null, + status: TrackStatusEnum.New, }); - await this.update(id, { ...track, status: TrackStatusEnum.New }); } async findOnYoutube(track: TrackEntity): Promise { - if (!(await this.get(track.id))) { + const dbTrack = await this.get(track.id); + + if (!dbTrack) { + this.logger.warn(`Skipping search for missing track ${track.id}`); return; } + + track = dbTrack; + await this.update(track.id, { ...track, status: TrackStatusEnum.Searching, @@ -114,9 +119,45 @@ export class TrackService { } await this.update(track.id, updatedTrack); - await this.trackDownloadQueue.add('', updatedTrack, { - jobId: `id-${updatedTrack.id}`, - }); + await this.enqueueDownload(updatedTrack.id); + } + + private async enqueueSearch(id: number): Promise { + const track = await this.get(id); + + if (!track) { + this.logger.warn(`Cannot enqueue search for missing track ${id}`); + return; + } + + const jobId = `search-${id}`; + const existingJob = await this.trackSearchQueue.getJob(jobId); + + if (existingJob) { + this.logger.warn(`Search job already exists for track ${id}`); + return; + } + + await this.trackSearchQueue.add('search-track', { id }, { jobId }); + } + + private async enqueueDownload(id: number): Promise { + const track = await this.get(id); + + if (!track) { + this.logger.warn(`Cannot enqueue download for missing track ${id}`); + return; + } + + const jobId = `download-${id}`; + const existingJob = await this.trackDownloadQueue.getJob(jobId); + + if (existingJob) { + this.logger.warn(`Download job already exists for track ${id}`); + return; + } + + await this.trackDownloadQueue.add('download-track', { id }, { jobId }); } async downloadFromYoutube(track: TrackEntity): Promise {