fix: addressed timing race condition

Author: jahilldev

api/src/queue/queue.service.ts

@@ -92,6 +92,14 @@ export class QueueService {
     });
   }
 
+  /** Refresh the lock timestamp for a task still owned by this worker. No-op if the task was reclaimed by another worker or is no longer RUNNING. */
+  async refreshLock(taskId: string, workerId: string): Promise<void> {
+    await this.prisma.queueTask.updateMany({
+      where: { id: taskId, lockedBy: workerId, status: 'RUNNING' },
+      data: { lockedAt: new Date() },
+    });
+  }
+
   /**
    * Mark a task failed. Retries with exponential backoff while attempts remain;
    * otherwise settles as FAILED. Returns true if it will be retried.

api/src/worker/worker.service.ts

@@ -78,6 +78,11 @@ export class WorkerService implements OnModuleInit, OnModuleDestroy {
 
   private async process(task: QueueTask): Promise<void> {
     this.logger.log(`Processing task ${task.id} (${task.kind}) for job ${task.jobId}`);
+    const heartbeatMs = Math.floor(this.config.get('QUEUE_LOCK_TTL_MS') / 3);
+    const heartbeat = setInterval(
+      () => void this.queue.refreshLock(task.id, WORKER_ID),
+      heartbeatMs,
+    );
     try {
       await this.orchestrator.processTask(task);
       await this.queue.complete(task.id);
@@ -91,6 +96,8 @@ export class WorkerService implements OnModuleInit, OnModuleDestroy {
             this.logger.error(`failed to escalate exhausted task: ${(err as Error).message}`),
           );
       }
+    } finally {
+      clearInterval(heartbeat);
     }
   }