diff --git a/server/services/roundReferenceAudioImport.js b/server/services/roundReferenceAudioImport.js index b86a6b015d..e1c0fc507d 100644 --- a/server/services/roundReferenceAudioImport.js +++ b/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); } })(); diff --git a/server/services/roundReferenceAudioImport.test.js b/server/services/roundReferenceAudioImport.test.js index b8bf58a6b0..0c46deebf3 100644 --- a/server/services/roundReferenceAudioImport.test.js +++ b/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(); + }); +}); diff --git a/server/services/trackYoutubeImport.js b/server/services/trackYoutubeImport.js index eca9ed32d9..bd1e8ffda1 100644 --- a/server/services/trackYoutubeImport.js +++ b/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); } })(); diff --git a/server/services/trackYoutubeImport.test.js b/server/services/trackYoutubeImport.test.js index afeac30725..4eb6fa898d 100644 --- a/server/services/trackYoutubeImport.test.js +++ b/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(); + }); +});