0.12.0-3 Plugin Integration

backend/src/services/pipelineService.js

@@ -4900,16 +4900,21 @@ class PipelineService extends EventEmitter {
       : (existing?.device && typeof existing.device === 'object'
           ? existing.device
           : { path: normalizedPath, mediaProfile: 'cd' });
+    const releasedDevice = {
+      ...fallbackDevice,
+      path: normalizedPath,
+      mediaProfile: 'cd'
+    };
 
     this._setCdDriveState(normalizedPath, {
       state: 'DISC_DETECTED',
       jobId: null,
-      device: fallbackDevice,
+      device: releasedDevice,
       progress: 0,
       eta: null,
       statusText: null,
       context: {
-        device: fallbackDevice,
+        device: releasedDevice,
         devicePath: normalizedPath,
         mediaProfile: 'cd'
       }
@@ -5143,6 +5148,149 @@ class PipelineService extends EventEmitter {
     return deleted;
   }
 
+  _forceUnlockDriveLocksForDeletedJobs(devicePaths = [], deletedJobIds = []) {
+    const normalizedPaths = Array.from(new Set(
+      (Array.isArray(devicePaths) ? devicePaths : [])
+        .map((value) => this.normalizeDrivePath(value))
+        .filter((value) => Boolean(value) && !value.startsWith('__virtual__'))
+    ));
+    if (normalizedPaths.length === 0 || typeof diskDetectionService.forceUnlockDevice !== 'function') {
+      return [];
+    }
+
+    const deletedSet = new Set(
+      (Array.isArray(deletedJobIds) ? deletedJobIds : [])
+        .map((value) => Number(value))
+        .filter((value) => Number.isFinite(value) && value > 0)
+        .map((value) => Math.trunc(value))
+    );
+    const activeLocks = diskDetectionService.getActiveLocks();
+    const forceUnlockedPaths = [];
+
+    for (const devicePath of normalizedPaths) {
+      const lock = activeLocks.find((entry) => this.normalizeDrivePath(entry?.path) === devicePath) || null;
+      if (!lock) {
+        continue;
+      }
+      const owners = Array.isArray(lock?.owners) ? lock.owners : [];
+      const ownerJobIds = owners
+        .map((owner) => Number(owner?.jobId))
+        .filter((value) => Number.isFinite(value) && value > 0)
+        .map((value) => Math.trunc(value));
+      const hasForeignOwner = ownerJobIds.some((jobId) => !deletedSet.has(jobId));
+      if (hasForeignOwner) {
+        logger.warn('drive-lock:force-unlock:skipped-foreign-owner', {
+          devicePath,
+          ownerJobIds
+        });
+        continue;
+      }
+      const clearedCount = Number(diskDetectionService.forceUnlockDevice(devicePath, {
+        reason: 'job_deleted',
+        deletedJobIds: Array.from(deletedSet)
+      }) || 0);
+      if (clearedCount > 0) {
+        forceUnlockedPaths.push(devicePath);
+      }
+    }
+
+    if (forceUnlockedPaths.length > 0) {
+      logger.warn('drive-lock:force-unlocked-after-delete', {
+        devicePaths: forceUnlockedPaths
+      });
+    }
+    return forceUnlockedPaths;
+  }
+
+  async _rescanDrivesAfterDelete(devicePaths = []) {
+    const normalizedPaths = Array.from(new Set(
+      (Array.isArray(devicePaths) ? devicePaths : [])
+        .map((value) => this.normalizeDrivePath(value))
+        .filter((value) => Boolean(value) && !value.startsWith('__virtual__'))
+    ));
+    if (normalizedPaths.length === 0) {
+      return;
+    }
+
+    const retryDelaysMs = [0, 450, 1250];
+    for (const devicePath of normalizedPaths) {
+      let lastResult = null;
+      for (let attempt = 0; attempt < retryDelaysMs.length; attempt += 1) {
+        const waitMs = retryDelaysMs[attempt];
+        if (waitMs > 0) {
+          await new Promise((resolve) => setTimeout(resolve, waitMs));
+        }
+        try {
+          lastResult = await diskDetectionService.rescanDriveAndEmit(devicePath);
+        } catch (error) {
+          logger.warn('job-delete:drive-rescan:failed', {
+            devicePath,
+            attempt: attempt + 1,
+            error: errorToMeta(error)
+          });
+          continue;
+        }
+
+        const detectedProfile = String(lastResult?.device?.mediaProfile || '').trim().toLowerCase();
+        const present = Boolean(lastResult?.present);
+        const locked = Boolean(lastResult?.locked);
+        if (locked) {
+          continue;
+        }
+        if (!present) {
+          continue;
+        }
+        if (detectedProfile && detectedProfile !== 'other' && detectedProfile !== 'unknown') {
+          break;
+        }
+      }
+      logger.info('job-delete:drive-rescan:done', {
+        devicePath,
+        emitted: lastResult?.emitted || 'none',
+        present: Boolean(lastResult?.present),
+        mediaProfile: lastResult?.device?.mediaProfile || null,
+        locked: Boolean(lastResult?.locked)
+      });
+    }
+  }
+
+  _forceStopActiveProcessForJob(jobId, options = {}) {
+    const normalizedJobId = Number(jobId);
+    if (!Number.isFinite(normalizedJobId) || normalizedJobId <= 0) {
+      return false;
+    }
+
+    const targetJobId = Math.trunc(normalizedJobId);
+    const processHandle = this.activeProcesses.get(targetJobId) || null;
+    if (!processHandle) {
+      return false;
+    }
+
+    const childPid = Number(processHandle?.child?.pid);
+    try {
+      processHandle?.cancel?.();
+    } catch (_error) {
+      // ignore cancellation race errors
+    }
+
+    if (Number.isFinite(childPid) && childPid > 0) {
+      try { process.kill(-childPid, 'SIGKILL'); } catch (_error) { /* noop */ }
+      try { process.kill(childPid, 'SIGKILL'); } catch (_error) { /* noop */ }
+    }
+    try {
+      processHandle?.child?.kill?.('SIGKILL');
+    } catch (_error) {
+      // noop
+    }
+
+    logger.warn('job-delete:active-process-killed', {
+      jobId: targetJobId,
+      reason: String(options?.reason || '').trim() || null,
+      pid: Number.isFinite(childPid) && childPid > 0 ? childPid : null
+    });
+    return true;
+  }
+
   async onJobsDeleted(jobIds = [], options = {}) {
     const normalizedIds = Array.isArray(jobIds)
       ? jobIds
@@ -5167,7 +5315,8 @@ class PipelineService extends EventEmitter {
     const removedQueueEntries = previousQueueLength - this.queueEntries.length;
 
     for (const jobId of normalizedIds) {
-      this.cancelRequestedByJob.delete(jobId);
+      this.cancelRequestedByJob.add(jobId);
+      this._forceStopActiveProcessForJob(jobId, { reason: 'job_deleted' });
       const lockForJob = this._getDriveLockByJobId(jobId);
       if (lockForJob?.devicePath) {
         affectedDevicePaths.add(this.normalizeDrivePath(lockForJob.devicePath));
@@ -5190,6 +5339,14 @@ class PipelineService extends EventEmitter {
       }
       this.activeProcesses.delete(jobId);
     }
+    // Keep the cancellation marker briefly so any late async step in the old
+    // pipeline thread aborts before spawning follow-up processes.
+    for (const jobId of normalizedIds) {
+      setTimeout(() => {
+        this.cancelRequestedByJob.delete(jobId);
+      }, 120000);
+    }
+    const forceUnlockedPaths = this._forceUnlockDriveLocksForDeletedJobs(Array.from(affectedDevicePaths), normalizedIds);
 
     let driveResetChanged = false;
     if (resetDriveState) {
@@ -5222,6 +5379,10 @@ class PipelineService extends EventEmitter {
       await this.emitQueueChanged();
       void this.pumpQueue();
     }
+
+    if (resetDriveState || forceUnlockedPaths.length > 0) {
+      void this._rescanDrivesAfterDelete(Array.from(affectedDevicePaths));
+    }
     return {
       cleaned: normalizedIds.length,
       removedQueueEntries
@@ -12780,8 +12941,9 @@ class PipelineService extends EventEmitter {
         applyLine(line, true);
       }
     });
-
-    this.activeProcesses.set(Number(normalizedJobId), processHandle);
+    const normalizedProcessKey = Number(normalizedJobId);
+    const previousProcessHandle = this.activeProcesses.get(normalizedProcessKey) || null;
+    this.activeProcesses.set(normalizedProcessKey, processHandle);
     this.syncPrimaryActiveProcess();
 
     try {
@@ -12814,8 +12976,18 @@ class PipelineService extends EventEmitter {
       logger.error('command:failed', { jobId, stage, source, error: errorToMeta(error) });
       throw error;
     } finally {
-      this.activeProcesses.delete(Number(normalizedJobId));
-      this.cancelRequestedByJob.delete(Number(normalizedJobId));
+      const activeHandleForJob = this.activeProcesses.get(normalizedProcessKey) || null;
+      if (activeHandleForJob === processHandle) {
+        if (previousProcessHandle && previousProcessHandle !== processHandle) {
+          this.activeProcesses.set(normalizedProcessKey, previousProcessHandle);
+        } else {
+          this.activeProcesses.delete(normalizedProcessKey);
+        }
+      }
+      // Preserve cancellation marker for outer/orchestrating handles (e.g. CD shared handle).
+      if (!(previousProcessHandle && previousProcessHandle !== processHandle)) {
+        this.cancelRequestedByJob.delete(normalizedProcessKey);
+      }
       this.syncPrimaryActiveProcess();
       await historyService.closeProcessLog(jobId);
       await this.emitQueueChanged();
@@ -14963,6 +15135,8 @@ class PipelineService extends EventEmitter {
         }
       }
     };
+    let currentTrackPosition = null;
+    let buildLiveContext = () => null;
     this.activeProcesses.set(processKey, sharedHandle);
     this.syncPrimaryActiveProcess();
 
@@ -14995,8 +15169,8 @@ class PipelineService extends EventEmitter {
       let encodeCompletedCount = 0;
       let currentPhase = 'rip';
       let currentTrackIndex = effectiveTrackTotal > 0 ? 1 : 0;
-      let currentTrackPosition = liveTrackRows[0]?.position || null;
-      const buildLiveContext = (failedTrackPosition = null) => buildCdLiveProgressSnapshot({
+      currentTrackPosition = liveTrackRows[0]?.position || null;
+      buildLiveContext = (failedTrackPosition = null) => buildCdLiveProgressSnapshot({
         trackRows: liveTrackRows,
         phase: currentPhase,
         trackIndex: currentTrackIndex,