From 0a418e773f35cee03158ad0715eefd844fbf8ed1 Mon Sep 17 00:00:00 2001 From: Giancarmine Salucci Date: Thu, 9 Jul 2026 03:59:52 +0200 Subject: [PATCH] D3: wire health preflight into pipeline before whisper submission Add health preflight check in runJob() after ensureWhisperRunning() and before submitJob(). Uses existing getHealth() from whisper.ts with retryWithBackoff() for transient failures. - Healthy whisper (status 'ok') -> job proceeds to submission - Unreachable whisper -> job fails early with 'whisper health check failed: ...' - Non-ok status (e.g. 'starting') -> job fails early with actual status - getHealth() 5000ms timeout preserved, no new HTTP client Tests: health-preflight.test.ts covers error message contracts and pipeline integration (YouTube + upload source) --- src/lib/server/pipeline.ts | 28 +++- src/tests/health-preflight.test.ts | 228 +++++++++++++++++++++++++++++ 2 files changed, 255 insertions(+), 1 deletion(-) create mode 100644 src/tests/health-preflight.test.ts diff --git a/src/lib/server/pipeline.ts b/src/lib/server/pipeline.ts index 6dcccba..3c2686e 100644 --- a/src/lib/server/pipeline.ts +++ b/src/lib/server/pipeline.ts @@ -1,8 +1,9 @@ import { createJob, updateJob, setJobStatus, getJob, resetJob } from './db.js'; import { downloadYouTube, saveUploadedFile, cleanupJobTmp, getUploadPath, cleanupUploadDir } from './downloader.js'; import { prepareAudio, cleanup as cleanupFiles } from './audio.js'; -import { submitJob, streamJob } from './whisper.js'; +import { submitJob, streamJob, getHealth } from './whisper.js'; import { ensureWhisperRunning } from './docker.js'; +import { retryWithBackoff } from './retry.js'; import type { AudioMode, Segment } from '$lib/types.js'; const WEBHOOK_BASE_URL = process.env.WEBHOOK_BASE_URL ?? 'http://localhost:3000'; @@ -151,6 +152,31 @@ async function runJob( // ── 3. Ensure whisper is running ────────────────────────────────────── await ensureWhisperRunning(); + // ── 3b. Health preflight: verify whisper reachability and status ────── + await retryWithBackoff( + async () => { + try { + const health = await getHealth(); + if (health.status !== 'ok') { + throw new Error(`whisper health check failed: status is "${health.status}"`); + } + } catch (err) { + if (err instanceof Error && err.message.startsWith('whisper health check failed')) { + throw err; + } + throw new Error(`whisper health check failed: ${err instanceof Error ? err.message : String(err)}`); + } + }, + { + maxAttempts: 3, + baseDelayMs: 1000, + onAttempt: (err, attempt, delayMs) => { + console.warn(`[pipeline] whisper health check attempt ${attempt} failed:`, err); + emitProgress(jobId, { type: 'warning', message: `health check attempt ${attempt} failed, retrying in ${delayMs}ms` }); + } + } + ); + // ── 4. Submit to whisper with webhook ──────────────────────────────── setJobStatus(jobId, 'transcribing', 10); emitProgress(jobId, { type: 'status', status: 'transcribing', progress: 10 }); diff --git a/src/tests/health-preflight.test.ts b/src/tests/health-preflight.test.ts new file mode 100644 index 0000000..e2b01f1 --- /dev/null +++ b/src/tests/health-preflight.test.ts @@ -0,0 +1,228 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; + +// ── Mock node-fetch so we control getHealth responses ────────────────────── +const mocks = vi.hoisted(() => ({ + fetch: vi.fn() +})); + +vi.mock('node-fetch', () => ({ default: mocks.fetch })); + +// ── Mock expensive / side-effectful pipeline dependencies ───────────────── +vi.mock('$lib/server/docker.js', () => ({ + ensureWhisperRunning: vi.fn().mockResolvedValue(undefined) +})); + +vi.mock('$lib/server/downloader.js', () => ({ + saveUploadedFile: vi.fn().mockResolvedValue('/tmp/fake.webm'), + downloadYouTube: vi.fn().mockResolvedValue({ + type: 'audio', + audioPath: '/tmp/fake.mp3', + title: 'Mocked Title' + }), + cleanupJobTmp: vi.fn().mockResolvedValue(undefined), + cleanupUploadDir: vi.fn().mockResolvedValue(undefined), + getUploadPath: vi.fn().mockReturnValue('/tmp/fake.webm') +})); + +vi.mock('$lib/server/audio.js', () => ({ + prepareAudio: vi.fn().mockResolvedValue({ + wavPath: '/tmp/fake.wav', + analysis: { meanVolume: -20, maxVolume: -10 } + }), + cleanup: vi.fn().mockResolvedValue(undefined) +})); + +vi.mock('$lib/server/remux.js', () => ({ + remuxUpload: vi.fn().mockResolvedValue('/tmp/fake-converted.webm') +})); + +// ── Imports after mocks ──────────────────────────────────────────────────── +import { retryWithBackoff } from '$lib/server/retry.js'; +import { getHealth } from '$lib/server/whisper.js'; +import { startYouTubeJob, startUploadJob } from '$lib/server/pipeline.js'; +import { getJob } from '$lib/server/db.js'; + +beforeEach(() => { + mocks.fetch.mockReset(); +}); + +afterEach(() => { + vi.useRealTimers(); +}); + +// ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ +// Unit: health preflight error-message contracts +// ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ + +describe('health preflight — error message contracts', () => { + it('wraps network error with "whisper health check failed:" prefix', async () => { + mocks.fetch.mockRejectedValue(new Error('connect ECONNREFUSED 127.0.0.1:8080')); + + await expect( + retryWithBackoff( + async () => { + try { + await getHealth(); + } catch (err) { + throw new Error( + `whisper health check failed: ${err instanceof Error ? err.message : String(err)}` + ); + } + }, + { maxAttempts: 1, baseDelayMs: 1 } + ) + ).rejects.toThrow(/whisper health check failed/); + }); + + it('throws error containing actual status when status !== "ok"', async () => { + mocks.fetch.mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ status: 'starting', model_state: 'loading' }) + }); + + await expect( + retryWithBackoff( + async () => { + const health = await getHealth(); + if (health.status !== 'ok') { + throw new Error(`whisper health check failed: status is "${health.status}"`); + } + }, + { maxAttempts: 1, baseDelayMs: 1 } + ) + ).rejects.toThrow(/whisper health check failed: status is "starting"/); + }); + + it('passes silently when health returns status "ok"', async () => { + mocks.fetch.mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ status: 'ok', model_state: 'loaded' }) + }); + + await expect( + retryWithBackoff( + async () => { + const health = await getHealth(); + if (health.status !== 'ok') { + throw new Error(`whisper health check failed: status is "${health.status}"`); + } + }, + { maxAttempts: 1, baseDelayMs: 1 } + ) + ).resolves.toBeUndefined(); + }); + + it('calls getHealth() (not a new fetch-based check)', async () => { + mocks.fetch.mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ status: 'ok' }) + }); + + const health = await getHealth(); + expect(health.status).toBe('ok'); + + // Verify the URL called is /health + const callUrl = mocks.fetch.mock.calls[0]?.[0] as string; + expect(callUrl).toMatch(/\/health$/); + }); +}); + +// ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ +// Integration: preflight wired into pipeline.runJob +// ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ + +describe('health preflight — pipeline integration', () => { + it('fails job with health error when whisper unreachable', async () => { + vi.useFakeTimers(); + mocks.fetch.mockRejectedValue(new Error('connect ECONNREFUSED')); + + const jobId = await startYouTubeJob('https://example.com/unreachable', 'auto'); + + // Advance past all retry delays (3 attempts, exp backoff ~1s+2s+4s ≈ 7-10s) + await vi.advanceTimersByTimeAsync(15000); + + await vi.waitFor( + () => { + const job = getJob(jobId); + expect(job).not.toBeNull(); + expect(job!.status).toBe('failed'); + expect(job!.error).toMatch(/whisper health check failed/); + }, + { timeout: 5000, interval: 50 } + ); + + vi.useRealTimers(); + }); + + it('fails job with health error including actual non-ok status', async () => { + vi.useFakeTimers(); + mocks.fetch.mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ status: 'starting', model_state: 'loading' }) + }); + + const jobId = await startYouTubeJob('https://example.com/starting', 'auto'); + + await vi.advanceTimersByTimeAsync(15000); + + await vi.waitFor( + () => { + const job = getJob(jobId); + expect(job).not.toBeNull(); + expect(job!.status).toBe('failed'); + expect(job!.error).toMatch(/whisper health check failed/); + expect(job!.error).toContain('starting'); + }, + { timeout: 5000, interval: 50 } + ); + + vi.useRealTimers(); + }); + + it('does NOT fail job with health error when whisper is healthy', async () => { + vi.useFakeTimers(); + mocks.fetch.mockResolvedValue({ + ok: true, + json: () => Promise.resolve({ status: 'ok', model_state: 'loaded' }) + }); + + const jobId = await startYouTubeJob('https://example.com/healthy', 'auto'); + + await vi.advanceTimersByTimeAsync(5000); + + // The job may be in any state (transcribing, done, or failed from a + // later stage) — the key assertion is it's NOT failed with a health error. + const job = getJob(jobId); + expect(job).not.toBeNull(); + if (job!.status === 'failed' && job!.error) { + expect(job!.error).not.toMatch(/whisper health check failed/); + } + + vi.useRealTimers(); + }); + + it('fails upload-source job when whisper unhealthy', async () => { + vi.useFakeTimers(); + mocks.fetch.mockRejectedValue(new Error('timeout')); + + const jobId = await startUploadJob( + Buffer.from('fake audio data'), + 'test-upload.webm', + 'auto' + ); + + await vi.advanceTimersByTimeAsync(15000); + + await vi.waitFor( + () => { + const job = getJob(jobId); + expect(job).not.toBeNull(); + expect(job!.status).toBe('failed'); + expect(job!.error).toMatch(/whisper health check failed/); + }, + { timeout: 5000, interval: 50 } + ); + + vi.useRealTimers(); + }); +});