From f90c74fe6ea0316a7d8d4d777ef2472326b917e0 Mon Sep 17 00:00:00 2001 From: WuYongJie Date: Mon, 25 May 2026 16:16:29 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=BC=BA=E8=A7=86=E9=A2=91=E8=BD=AC?= =?UTF-8?q?=E7=A0=81=E6=81=A2=E5=A4=8D=E4=B8=8E=E5=A4=B1=E8=B4=A5=E5=A4=84?= =?UTF-8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lib/__tests__/video-transcode-queue.test.ts | 34 +++++++++ lib/__tests__/video-transcode-service.test.ts | 75 +++++++++++++++++++ lib/video-processor.ts | 12 +-- lib/video-transcode-queue.ts | 8 +- lib/video-transcode-service.ts | 23 +++++- 5 files changed, 137 insertions(+), 15 deletions(-) create mode 100644 lib/__tests__/video-transcode-queue.test.ts diff --git a/lib/__tests__/video-transcode-queue.test.ts b/lib/__tests__/video-transcode-queue.test.ts new file mode 100644 index 0000000..1392af3 --- /dev/null +++ b/lib/__tests__/video-transcode-queue.test.ts @@ -0,0 +1,34 @@ +export {} + +describe('video-transcode-queue', () => { + const originalEnv = process.env + + afterEach(() => { + process.env = originalEnv + jest.resetModules() + }) + + it('无参创建单例时会读取 VIDEO_TRANSCODE_MAX_CONCURRENT', async () => { + process.env = { + ...originalEnv, + VIDEO_TRANSCODE_MAX_CONCURRENT: '3', + } + jest.resetModules() + + const { transcodeQueue } = await import('../video-transcode-queue') + + expect(transcodeQueue.getStatus().maxConcurrent).toBe(3) + }) + + it('非法并发配置会回退为 1，保护低性能私有化环境', async () => { + process.env = { + ...originalEnv, + VIDEO_TRANSCODE_MAX_CONCURRENT: '0', + } + jest.resetModules() + + const { transcodeQueue } = await import('../video-transcode-queue') + + expect(transcodeQueue.getStatus().maxConcurrent).toBe(1) + }) +}) diff --git a/lib/__tests__/video-transcode-service.test.ts b/lib/__tests__/video-transcode-service.test.ts index b9b3562..8dd9f58 100644 --- a/lib/__tests__/video-transcode-service.test.ts +++ b/lib/__tests__/video-transcode-service.test.ts @@ -35,6 +35,8 @@ describe('video-transcode-service', () => { jest.resetAllMocks() jest.useRealTimers() transcodeQueue.add.mockResolvedValue(undefined) + prisma.videoTranscodeJob.upsert.mockResolvedValue({}) + prisma.material.update.mockResolvedValue({}) }) afterEach(() => { @@ -105,6 +107,35 @@ describe('video-transcode-service', () => { expect(prisma.$queryRaw.mock.calls[0][0].join('')).toContain(`OR m."materialType" = '视频'`) }) + it('恢复批次达到上限时会标记 hasMore 并安排继续恢复', async () => { + jest.useFakeTimers() + const { recoverStaleVideoTranscodeJobs } = require('../video-transcode-service') + + prisma.videoTranscodeJob.findMany.mockResolvedValue( + Array.from({ length: 20 }, (_, index) => ({ + materialId: `material-${index}`, + videoSourceUrl: `s3/video-${index}.mp4`, + material: { + id: `material-${index}`, + videoSourceUrl: `s3/video-${index}.mp4`, + videoCoverUrl: null, + }, + })) + ) + prisma.$queryRaw.mockResolvedValue([]) + + const result = await recoverStaleVideoTranscodeJobs() + + expect(result).toEqual({ recovered: 20, hasMore: true }) + expect(transcodeQueue.add).toHaveBeenCalledTimes(20) + + prisma.videoTranscodeJob.findMany.mockResolvedValue([]) + prisma.$queryRaw.mockResolvedValue([]) + await jest.advanceTimersByTimeAsync(1000) + + expect(prisma.videoTranscodeJob.findMany).toHaveBeenCalledTimes(2) + }) + it('任务失败但还有重试次数时，会重新写回 queued 并延迟重新入队', async () => { jest.useFakeTimers() const { enqueueVideoTranscodeJob } = require('../video-transcode-service') @@ -154,4 +185,48 @@ describe('video-transcode-service', () => { expect(transcodeQueue.add).toHaveBeenCalledTimes(2) expect(transcodeQueue.add.mock.calls[1][0]).toBe('material-1') }) + + it('任务开始处理时同步更新素材状态为 processing', async () => { + const { enqueueVideoTranscodeJob } = require('../video-transcode-service') + + prisma.videoTranscodeJob.update.mockResolvedValue({ + materialId: 'material-1', + attempts: 1, + maxAttempts: 3, + }) + processVideoTranscode.mockResolvedValue({ + preview540Url: 's3/preview.mp4', + clipUrl: 's3/clip.mp4', + coverUrl: null, + duration: 12, + }) + + let queueRun: Promise | undefined + transcodeQueue.add.mockImplementationOnce(async ( + materialId: string, + videoSourceUrl: string, + needsCoverImage: boolean, + taskFn: (materialId: string, videoSourceUrl: string, needsCoverImage: boolean) => Promise + ) => { + queueRun = taskFn(materialId, videoSourceUrl, needsCoverImage) + await queueRun + }) + + await enqueueVideoTranscodeJob({ + materialId: 'material-1', + videoSourceUrl: 's3/video.mp4', + needsCoverImage: false, + }) + + await queueRun + + expect(prisma.material.update).toHaveBeenCalledWith({ + where: { id: 'material-1' }, + data: { videoProcessingStatus: 'processing' }, + }) + expect(prisma.material.update).toHaveBeenLastCalledWith({ + where: { id: 'material-1' }, + data: expect.objectContaining({ videoProcessingStatus: 'done' }), + }) + }) }) diff --git a/lib/video-processor.ts b/lib/video-processor.ts index 2695d82..a32a827 100644 --- a/lib/video-processor.ts +++ b/lib/video-processor.ts @@ -89,8 +89,8 @@ export async function checkFFmpegAvailable(): Promise { * 从 S3 下载视频到临时文件 */ async function downloadVideoToTemp(s3Url: string, tempPath: string): Promise { - if (!isS3Url(s3Url)) { - throw new Error('Invalid S3 URL') + if (!s3Url) { + throw new Error('Invalid S3 path') } const stream = await getS3ObjectStream(s3Url) @@ -225,13 +225,7 @@ export async function processVideoTranscode( // 检查 ffmpeg 是否可用 const ffmpegAvailable = await checkFFmpegAvailable() if (!ffmpegAvailable) { - console.warn('[VideoProcessor] FFmpeg not available, skipping transcode') - return { - preview540Url: null, - clipUrl: null, - coverUrl: null, - duration: null, - } + throw new Error('FFmpeg 不可用，无法执行视频转码') } const tempDir = tmpdir() diff --git a/lib/video-transcode-queue.ts b/lib/video-transcode-queue.ts index a3d5787..c686c18 100644 --- a/lib/video-transcode-queue.ts +++ b/lib/video-transcode-queue.ts @@ -18,10 +18,14 @@ class VideoTranscodeQueue { private queued: Set = new Set() private maxConcurrent: number - constructor(maxConcurrent: number = 1) { + constructor(maxConcurrent?: number) { // 默认并发数为 1，适合低性能 NAS 设备 // 可以通过环境变量配置：VIDEO_TRANSCODE_MAX_CONCURRENT - this.maxConcurrent = maxConcurrent || parseInt(process.env.VIDEO_TRANSCODE_MAX_CONCURRENT || '1', 10) + const configuredMaxConcurrent = parseInt(process.env.VIDEO_TRANSCODE_MAX_CONCURRENT || '1', 10) + const resolvedMaxConcurrent = maxConcurrent ?? configuredMaxConcurrent + this.maxConcurrent = Number.isFinite(resolvedMaxConcurrent) && resolvedMaxConcurrent > 0 + ? resolvedMaxConcurrent + : 1 } /** diff --git a/lib/video-transcode-service.ts b/lib/video-transcode-service.ts index 84189af..76e0af3 100644 --- a/lib/video-transcode-service.ts +++ b/lib/video-transcode-service.ts @@ -7,6 +7,7 @@ const DEFAULT_MAX_ATTEMPTS = 3 const DEFAULT_LOCK_TIMEOUT_MS = 30 * 60 * 1000 const DEFAULT_RETRY_DELAY_MS = 5 * 1000 const MAX_RETRY_DELAY_MS = 30 * 1000 +const RECOVERY_BATCH_SIZE = 20 function getMaxAttempts() { const retries = parseInt(process.env.VIDEO_TRANSCODE_MAX_RETRIES || '2', 10) @@ -45,7 +46,7 @@ async function runVideoTranscodeTask( ) { const maxAttempts = getMaxAttempts() - const job = await (prisma as any).videoTranscodeJob.update({ + const job = await (prisma as any).videoTranscodeJob.update({ where: { materialId }, data: { status: 'processing', @@ -61,6 +62,11 @@ async function runVideoTranscodeTask( try { console.log(`[Transcode] Starting persistent task for material ${materialId} (attempt ${job.attempts}/${maxAttempts})`) + await prisma.material.update({ + where: { id: materialId }, + data: { videoProcessingStatus: 'processing' }, + }) + const result = await processVideoTranscode({ inputPath: videoSourceUrl, outputDir: tmpdir(), @@ -211,7 +217,7 @@ export async function recoverStaleVideoTranscodeJobs() { }, }, }, - take: 20, + take: RECOVERY_BATCH_SIZE, orderBy: { updatedAt: 'asc' }, }) @@ -228,7 +234,7 @@ export async function recoverStaleVideoTranscodeJobs() { AND m."videoProcessingStatus" IN ('pending', 'processing') AND j.id IS NULL ORDER BY m."updatedAt" ASC - LIMIT 20 + LIMIT ${RECOVERY_BATCH_SIZE} ` let recovered = 0 @@ -272,5 +278,14 @@ export async function recoverStaleVideoTranscodeJobs() { console.log(`[Transcode] Recovered ${recovered} queued/stale video transcode job(s)`) } - return { recovered } + const hasMore = staleJobs.length >= RECOVERY_BATCH_SIZE || orphanMaterials.length >= RECOVERY_BATCH_SIZE + if (hasMore) { + setTimeout(() => { + recoverStaleVideoTranscodeJobs().catch((error) => { + console.error('[Transcode] Continued recovery failed:', error) + }) + }, 1000) + } + + return { recovered, hasMore } }