Harden queue execution and job lifecycle

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<string>(EnvironmentEnum.REDIS_HOST),

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<TrackEntity, void>): Promise<void> {
+    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);
   }

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<TrackEntity, void, string>): Promise<void> {
+    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);
   }
 }

src/backend/src/track/track.service.ts

@@ -58,9 +58,7 @@ export class TrackService {
 
   async create(track: TrackEntity, playlist?: PlaylistEntity): Promise<void> {
     const savedTrack = await this.repository.save({ ...track, playlist });
-    await this.trackSearchQueue.add('', savedTrack, {
+    await this.enqueueSearch(savedTrack.id);
-      jobId: `id-${savedTrack.id}`,
-    });
     this.io.emit(WsTrackOperation.New, {
       track: savedTrack,
       playlistId: playlist.id,
@@ -74,23 +72,30 @@ export class TrackService {
 
   async retry(id: number): Promise<void> {
     const track = await this.get(id);
-    const existingJob = await this.trackSearchQueue.getJob(`id-${id}`);
 
-    if (existingJob) {
+    if (!track) {
-      this.logger.warn(`Search job already exists for track ${id}`);
+      this.logger.warn(`Cannot retry missing track ${id}`);
       return;
     }
 
-    await this.trackSearchQueue.add('', track, {
+    await this.enqueueSearch(id);
-      jobId: `id-${id}`,
+    await this.update(id, {
+      ...track,
+      error: null,
+      status: TrackStatusEnum.New,
     });
-    await this.update(id, { ...track, status: TrackStatusEnum.New });
   }
 
   async findOnYoutube(track: TrackEntity): Promise<void> {
-    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, {
+    await this.enqueueDownload(updatedTrack.id);
-      jobId: `id-${updatedTrack.id}`,
+  }
-    });
+
+  private async enqueueSearch(id: number): Promise<void> {
+    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<void> {
+    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<void> {