fix: cancel yt-dlp audio imports reliably when no child process is running

server/services/roundReferenceAudioImport.js

@@ -26,17 +26,34 @@ import { resolveYtDlpBinaries, downloadAudioToTempMp3, cleanupYtDlpTemp } from '
 export const REFERENCE_AUDIO_MAX_BYTES = 60 * 1024 * 1024; // 60MB
 export const REFERENCE_AUDIO_MAX_DURATION_SEC = 20 * 60; // 20 minutes
 
-// jobId -> { clients, lastPayload, process }
+// jobId -> { clients, lastPayload, process, canceled }
 const importJobs = new Map();
 
 export const attachReferenceAudioSseClient = (jobId, res) => attachSse(importJobs, jobId, res);
 
-/** Cancel an in-flight import. Returns false if the job is unknown or already finished. */
+// The job map is module-private, which left cancelReferenceAudioImport
+// untestable. Exposing it lets a test register a fake job and actually run the
+// cancel — same rationale as videoDownload.js's __testing export.
+export const __testing = { importJobs };
+
+/**
+ * Cancel an in-flight import. Returns false if the job is unknown or already
+ * cancelled.
+ *
+ * `job.process` is transient — it is null while the job is awaiting setup and
+ * again during post-processing (runYtDlp clears it on exit) — so cancellation
+ * is recorded as a `job.canceled` flag the kickoff re-checks at each phase
+ * boundary, and the child is only signalled when one is actually running.
+ */
 export function cancelReferenceAudioImport(jobId) {
   const job = importJobs.get(jobId);
-  if (!job || !job.process) return false;
+  // The job lingers in the map after it ends (closeJobAfterDelay evicts it on
+  // a timer), so a terminal status — not just the flag — is what makes a cancel
+  // for an already-finished import report false.
+  if (!job || job.canceled || job.status !== 'running') return false;
+  job.canceled = true;
   const proc = job.process;
-  killWithEscalation(proc, { label: 'yt-dlp reference audio', stillRunning: () => job.process === proc });
+  if (proc) killWithEscalation(proc, { label: 'yt-dlp reference audio', stillRunning: () => job.process === proc });
   return true;
 }
 
@@ -56,12 +73,27 @@ export async function startReferenceAudioImport(url) {
 
   const jobId = randomUUID();
   const tempPrefix = `portos-refaudio-${jobId}`;
-  const job = { id: jobId, status: 'running', clients: [], process: null };
+  const job = { id: jobId, status: 'running', clients: [], process: null, canceled: false };
   importJobs.set(jobId, job);
   console.log(`🎧 Reference-audio import ${shortId(jobId)} — ${url}`);
 
   (async () => {
+    // Cancel can land in three windows: before the spawn, while yt-dlp runs
+    // (the core reports `canceled`), or during post-processing after the child
+    // has exited. Only the middle one is the core's to detect — `job.process`
+    // is null in the other two, so the flag is what carries them. The guard is
+    // re-run at each phase boundary rather than only after the download, so
+    // adding an await ahead of the spawn can't silently reopen window one.
+    const abortIfCanceled = async (canceled = job.canceled) => {
+      if (!canceled) return false;
+      console.log(`🛑 Reference-audio import ${shortId(jobId)} cancelled`);
+      broadcastSse(job, { type: 'canceled' });
+      await cleanupYtDlpTemp(tempPrefix);
+      return true;
+    };
+
     try {
+      if (await abortIfCanceled()) return;
       const result = await downloadAudioToTempMp3({
         url, ytDlp, ffmpeg, tempPrefix,
         maxBytes: REFERENCE_AUDIO_MAX_BYTES,
@@ -70,11 +102,10 @@ export async function startReferenceAudioImport(url) {
         registerProcess: (proc) => { job.process = proc; },
       });
 
-      if (result.outcome === 'canceled') {
-        console.log(`🛑 Reference-audio import ${shortId(jobId)} cancelled`);
-        broadcastSse(job, { type: 'canceled' });
-        return;
-      }
+      // The core reports `canceled` only for a kill we issued, so the two
+      // signals normally agree — check both so an externally-killed child
+      // still ends as a cancel rather than a failure.
+      if (await abortIfCanceled(job.canceled || result.outcome === 'canceled')) return;
       if (result.outcome === 'failed') {
         throw new Error(result.reason);
       }
@@ -93,6 +124,7 @@ export async function startReferenceAudioImport(url) {
       // the core handed off and no longer owns.
       await cleanupYtDlpTemp(tempPrefix);
     } finally {
+      job.status = 'done';
       closeJobAfterDelay(importJobs, jobId);
     }
   })();

server/services/roundReferenceAudioImport.test.js

@@ -22,7 +22,10 @@ const { assertPublicHttpUrl } = await import('../lib/safeUrlFetch.js');
 const { broadcastSse } = await import('../lib/sseUtils.js');
 const { resolveYtDlpBinaries, downloadAudioToTempMp3, cleanupYtDlpTemp } = await import('./ytdlpAudioImport.js');
 const { importFileToUploads } = await import('../lib/fileUtils.js');
-const { startReferenceAudioImport } = await import('./roundReferenceAudioImport.js');
+const { killWithEscalation } = await import('../lib/killWithEscalation.js');
+const {
+  startReferenceAudioImport, cancelReferenceAudioImport, __testing,
+} = await import('./roundReferenceAudioImport.js');
 
 // Let the detached kickoff IIFE run to its terminal broadcast.
 const flush = async () => {
@@ -89,3 +92,61 @@ describe('startReferenceAudioImport — outcomes', () => {
     expect(importFileToUploads).not.toHaveBeenCalled();
   });
 });
+
+describe('cancelReferenceAudioImport', () => {
+  it('returns false for an unknown job id', () => {
+    expect(cancelReferenceAudioImport('nope')).toBe(false);
+    expect(killWithEscalation).not.toHaveBeenCalled();
+  });
+
+  it('records the cancel and returns true even when no child has spawned yet', () => {
+    const job = { id: 'j1', status: 'running', clients: [], process: null, canceled: false };
+    __testing.importJobs.set('j1', job);
+    expect(cancelReferenceAudioImport('j1')).toBe(true);
+    expect(job.canceled).toBe(true);
+    expect(killWithEscalation).not.toHaveBeenCalled();
+    __testing.importJobs.delete('j1');
+  });
+
+  it('returns false once the job has finished, while it lingers in the map', () => {
+    // closeJobAfterDelay evicts on a timer, so a finished job is still present —
+    // cancelling it must not report an accepted cancel to the client.
+    __testing.importJobs.set('j3', { id: 'j3', status: 'done', clients: [], process: null, canceled: false });
+    expect(cancelReferenceAudioImport('j3')).toBe(false);
+    __testing.importJobs.delete('j3');
+  });
+
+  it('signals a running child and refuses a second cancel', () => {
+    const proc = { pid: 4321 };
+    const job = { id: 'j2', status: 'running', clients: [], process: proc, canceled: false };
+    __testing.importJobs.set('j2', job);
+    expect(cancelReferenceAudioImport('j2')).toBe(true);
+    expect(killWithEscalation).toHaveBeenCalledWith(proc, expect.objectContaining({ label: 'yt-dlp reference audio' }));
+    expect(cancelReferenceAudioImport('j2')).toBe(false);
+    expect(killWithEscalation).toHaveBeenCalledOnce();
+    __testing.importJobs.delete('j2');
+  });
+});
+
+describe('startReferenceAudioImport — cancellation windows', () => {
+  it('emits canceled and persists nothing when cancelled after the download finished', async () => {
+    // The child has already exited by post-processing time (job.process is back
+    // to null), so only the canceled flag can carry the cancel.
+    // An assertion thrown inside the detached kickoff would be swallowed by its
+    // catch, so record the result and assert it out here.
+    let canceledOk = null;
+    downloadAudioToTempMp3.mockImplementationOnce(async () => {
+      const jobId = [...__testing.importJobs.keys()].pop();
+      canceledOk = cancelReferenceAudioImport(jobId);
+      return { outcome: 'complete', outPath: '/tmp/x.mp3', title: 'Clip' };
+    });
+
+    await startReferenceAudioImport('https://example.com/clip');
+    await flush();
+
+    expect(canceledOk).toBe(true);
+    expect(broadcastSse).toHaveBeenCalledWith(expect.anything(), { type: 'canceled' });
+    expect(importFileToUploads).not.toHaveBeenCalled();
+    expect(cleanupYtDlpTemp).toHaveBeenCalled();
+  });
+});

server/services/trackYoutubeImport.js

@@ -37,17 +37,34 @@ export function assertYoutubeUrl(url) {
   }
 }
 
-// jobId -> { clients, lastPayload, process }
+// jobId -> { clients, lastPayload, process, canceled }
 const importJobs = new Map();
 
 export const attachImportSseClient = (jobId, res) => attachSse(importJobs, jobId, res);
 
-/** Cancel an in-flight import. Returns false if the job is unknown or already finished. */
+// The job map is module-private, which left cancelYoutubeImport untestable.
+// Exposing it lets a test register a fake job and actually run the cancel —
+// same rationale as videoDownload.js's __testing export.
+export const __testing = { importJobs };
+
+/**
+ * Cancel an in-flight import. Returns false if the job is unknown or already
+ * cancelled.
+ *
+ * `job.process` is transient — it is null while the job is awaiting setup and
+ * again during post-processing (runYtDlp clears it on exit) — so cancellation
+ * is recorded as a `job.canceled` flag the kickoff re-checks at each phase
+ * boundary, and the child is only signalled when one is actually running.
+ */
 export function cancelYoutubeImport(jobId) {
   const job = importJobs.get(jobId);
-  if (!job || !job.process) return false;
+  // The job lingers in the map after it ends (closeJobAfterDelay evicts it on
+  // a timer), so a terminal status — not just the flag — is what makes a cancel
+  // for an already-finished import report false.
+  if (!job || job.canceled || job.status !== 'running') return false;
+  job.canceled = true;
   const proc = job.process;
-  killWithEscalation(proc, { label: 'yt-dlp import', stillRunning: () => job.process === proc });
+  if (proc) killWithEscalation(proc, { label: 'yt-dlp import', stillRunning: () => job.process === proc });
   return true;
 }
 
@@ -63,12 +80,27 @@ export async function startYoutubeImport(url) {
 
   const jobId = randomUUID();
   const tempPrefix = `portos-ytimport-${jobId}`;
-  const job = { id: jobId, status: 'running', clients: [], process: null };
+  const job = { id: jobId, status: 'running', clients: [], process: null, canceled: false };
   importJobs.set(jobId, job);
   console.log(`📺 YouTube import ${shortId(jobId)} — ${url}`);
 
   (async () => {
+    // Cancel can land in three windows: before the spawn, while yt-dlp runs
+    // (the core reports `canceled`), or during post-processing after the child
+    // has exited. Only the middle one is the core's to detect — `job.process`
+    // is null in the other two, so the flag is what carries them. The guard is
+    // re-run at each phase boundary rather than only after the download, so
+    // adding an await ahead of the spawn can't silently reopen window one.
+    const abortIfCanceled = async (canceled = job.canceled) => {
+      if (!canceled) return false;
+      console.log(`🛑 YouTube import ${shortId(jobId)} cancelled`);
+      broadcastSse(job, { type: 'canceled' });
+      await cleanupYtDlpTemp(tempPrefix);
+      return true;
+    };
+
     try {
+      if (await abortIfCanceled()) return;
       const result = await downloadAudioToTempMp3({
         url, ytDlp, ffmpeg, tempPrefix,
         maxBytes: MUSIC_UPLOAD_MAX_BYTES,
@@ -77,11 +109,10 @@ export async function startYoutubeImport(url) {
         registerProcess: (proc) => { job.process = proc; },
       });
 
-      if (result.outcome === 'canceled') {
-        console.log(`🛑 YouTube import ${shortId(jobId)} cancelled`);
-        broadcastSse(job, { type: 'canceled' });
-        return;
-      }
+      // The core reports `canceled` only for a kill we issued, so the two
+      // signals normally agree — check both so an externally-killed child
+      // still ends as a cancel rather than a failure.
+      if (await abortIfCanceled(job.canceled || result.outcome === 'canceled')) return;
       if (result.outcome === 'failed') {
         throw new Error(result.reason);
       }
@@ -102,6 +133,7 @@ export async function startYoutubeImport(url) {
       // produced outPath the core handed off and no longer owns.
       await cleanupYtDlpTemp(tempPrefix);
     } finally {
+      job.status = 'done';
       closeJobAfterDelay(importJobs, jobId);
     }
   })();

server/services/trackYoutubeImport.test.js

@@ -15,12 +15,29 @@ vi.mock('./tracks/index.js', () => ({
   DURATION_MAX_SEC: 3600,
 }));
 vi.mock('../lib/childProcess.js', async (importOriginal) => ({ ...(await importOriginal()), spawn: vi.fn() }));
+vi.mock('../lib/sseUtils.js', () => ({
+  broadcastSse: vi.fn(),
+  attachSseClient: vi.fn(() => true),
+  closeJobAfterDelay: vi.fn(),
+}));
+vi.mock('../lib/killWithEscalation.js', () => ({ killWithEscalation: vi.fn() }));
+// Spy that calls through, so the argv test still exercises the real yt-dlp core
+// while the cancellation tests can substitute an outcome.
+vi.mock('./ytdlpAudioImport.js', async (importOriginal) => {
+  const actual = await importOriginal();
+  return { ...actual, downloadAudioToTempMp3: vi.fn(actual.downloadAudioToTempMp3) };
+});
 
 const { findYtDlp } = await import('../lib/ytdlp.js');
 const { findFfmpeg } = await import('../lib/ffmpeg.js');
 const { spawn } = await import('../lib/childProcess.js');
+const { broadcastSse } = await import('../lib/sseUtils.js');
+const { killWithEscalation } = await import('../lib/killWithEscalation.js');
+const { downloadAudioToTempMp3 } = await import('./ytdlpAudioImport.js');
+const { importUploadedTrack } = await import('./pipeline/musicLibrary.js');
+const { createTrack } = await import('./tracks/index.js');
 const {
-  YOUTUBE_URL_RE, assertYoutubeUrl, startYoutubeImport,
+  YOUTUBE_URL_RE, assertYoutubeUrl, startYoutubeImport, cancelYoutubeImport, __testing,
 } = await import('./trackYoutubeImport.js');
 
 // A fake yt-dlp child that immediately closes with the given exit code —
@@ -107,3 +124,72 @@ describe('startYoutubeImport — yt-dlp argv', () => {
     expect(args).toEqual(expect.arrayContaining(['--match-filters', 'duration <= 3600']));
   });
 });
+
+describe('cancelYoutubeImport', () => {
+  it('returns false for an unknown job id', () => {
+    expect(cancelYoutubeImport('nope')).toBe(false);
+    expect(killWithEscalation).not.toHaveBeenCalled();
+  });
+
+  it('records the cancel and returns true even when no child has spawned yet', () => {
+    const job = { id: 'j1', status: 'running', clients: [], process: null, canceled: false };
+    __testing.importJobs.set('j1', job);
+    expect(cancelYoutubeImport('j1')).toBe(true);
+    expect(job.canceled).toBe(true);
+    // job.process is null during setup and again during post-processing — there
+    // is nothing to signal, but the cancel must still be recorded.
+    expect(killWithEscalation).not.toHaveBeenCalled();
+    __testing.importJobs.delete('j1');
+  });
+
+  it('returns false once the job has finished, while it lingers in the map', () => {
+    // closeJobAfterDelay evicts on a timer, so a finished job is still present —
+    // cancelling it must not report an accepted cancel to the client.
+    __testing.importJobs.set('j3', { id: 'j3', status: 'done', clients: [], process: null, canceled: false });
+    expect(cancelYoutubeImport('j3')).toBe(false);
+    __testing.importJobs.delete('j3');
+  });
+
+  it('signals a running child and refuses a second cancel', () => {
+    const proc = { pid: 1234 };
+    const job = { id: 'j2', status: 'running', clients: [], process: proc, canceled: false };
+    __testing.importJobs.set('j2', job);
+    expect(cancelYoutubeImport('j2')).toBe(true);
+    expect(killWithEscalation).toHaveBeenCalledWith(proc, expect.objectContaining({ label: 'yt-dlp import' }));
+    expect(cancelYoutubeImport('j2')).toBe(false);
+    expect(killWithEscalation).toHaveBeenCalledOnce();
+    __testing.importJobs.delete('j2');
+  });
+});
+
+describe('startYoutubeImport — cancellation windows', () => {
+  it('emits canceled and creates no track when cancelled after the download finished', async () => {
+    // Cancel lands while the download is settling: the child has already exited
+    // (job.process back to null), so only the flag can carry the cancel.
+    // An assertion thrown inside the detached kickoff would be swallowed by its
+    // catch, so record the result and assert it out here.
+    let canceledOk = null;
+    downloadAudioToTempMp3.mockImplementationOnce(async () => {
+      const jobId = [...__testing.importJobs.keys()].pop();
+      canceledOk = cancelYoutubeImport(jobId);
+      return { outcome: 'complete', outPath: '/tmp/x.mp3', title: 'Clip' };
+    });
+
+    await startYoutubeImport('https://youtu.be/dQw4w9WgXcQ');
+    for (let i = 0; i < 6; i += 1) await new Promise((r) => setImmediate(r));
+
+    expect(canceledOk).toBe(true);
+    expect(broadcastSse).toHaveBeenCalledWith(expect.anything(), { type: 'canceled' });
+    expect(importUploadedTrack).not.toHaveBeenCalled();
+    expect(createTrack).not.toHaveBeenCalled();
+  });
+
+  it('emits canceled when the core reports the child was killed mid-download', async () => {
+    downloadAudioToTempMp3.mockResolvedValueOnce({ outcome: 'canceled' });
+    await startYoutubeImport('https://youtu.be/dQw4w9WgXcQ');
+    for (let i = 0; i < 6; i += 1) await new Promise((r) => setImmediate(r));
+
+    expect(broadcastSse).toHaveBeenCalledWith(expect.anything(), { type: 'canceled' });
+    expect(createTrack).not.toHaveBeenCalled();
+  });
+});