兼容转码恢复启动环境
This commit is contained in:
parent
6a5077312e
commit
5b6dd8b487
@ -107,6 +107,26 @@ describe('video-transcode-service', () => {
|
||||
expect(prisma.$queryRaw.mock.calls[0][0].join('')).toContain(`OR m."materialType" = '视频'`)
|
||||
})
|
||||
|
||||
it('Prisma Client 缺少 videoTranscodeJob 时,启动恢复会安全跳过', async () => {
|
||||
const { recoverStaleVideoTranscodeJobs } = require('../video-transcode-service')
|
||||
const originalDelegate = prisma.videoTranscodeJob
|
||||
const warnSpy = jest.spyOn(console, 'warn').mockImplementation(() => {})
|
||||
|
||||
prisma.videoTranscodeJob = undefined
|
||||
|
||||
try {
|
||||
const result = await recoverStaleVideoTranscodeJobs()
|
||||
|
||||
expect(result).toEqual({ recovered: 0, hasMore: false, skipped: true })
|
||||
expect(prisma.$queryRaw).not.toHaveBeenCalled()
|
||||
expect(transcodeQueue.add).not.toHaveBeenCalled()
|
||||
expect(warnSpy).toHaveBeenCalledWith(expect.stringContaining('跳过启动恢复'))
|
||||
} finally {
|
||||
prisma.videoTranscodeJob = originalDelegate
|
||||
warnSpy.mockRestore()
|
||||
}
|
||||
})
|
||||
|
||||
it('恢复批次达到上限时会标记 hasMore 并安排继续恢复', async () => {
|
||||
jest.useFakeTimers()
|
||||
const { recoverStaleVideoTranscodeJobs } = require('../video-transcode-service')
|
||||
|
||||
@ -8,6 +8,32 @@ 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
|
||||
type VideoTranscodeJobDelegate = {
|
||||
upsert: (...args: any[]) => Promise<any>
|
||||
findMany: (...args: any[]) => Promise<any[]>
|
||||
update: (...args: any[]) => Promise<any>
|
||||
}
|
||||
|
||||
function getVideoTranscodeJobDelegate(): VideoTranscodeJobDelegate | null {
|
||||
const delegate = (prisma as any).videoTranscodeJob
|
||||
if (
|
||||
delegate &&
|
||||
typeof delegate.upsert === 'function' &&
|
||||
typeof delegate.findMany === 'function' &&
|
||||
typeof delegate.update === 'function'
|
||||
) {
|
||||
return delegate
|
||||
}
|
||||
return null
|
||||
}
|
||||
|
||||
function requireVideoTranscodeJobDelegate(): VideoTranscodeJobDelegate {
|
||||
const delegate = getVideoTranscodeJobDelegate()
|
||||
if (!delegate) {
|
||||
throw new Error('Prisma Client 缺少 videoTranscodeJob,请先执行数据库迁移并重新生成 Prisma Client')
|
||||
}
|
||||
return delegate
|
||||
}
|
||||
|
||||
function getMaxAttempts() {
|
||||
const retries = parseInt(process.env.VIDEO_TRANSCODE_MAX_RETRIES || '2', 10)
|
||||
@ -45,8 +71,9 @@ async function runVideoTranscodeTask(
|
||||
needsCoverImage: boolean
|
||||
) {
|
||||
const maxAttempts = getMaxAttempts()
|
||||
const videoTranscodeJob = requireVideoTranscodeJobDelegate()
|
||||
|
||||
const job = await (prisma as any).videoTranscodeJob.update({
|
||||
const job = await videoTranscodeJob.update({
|
||||
where: { materialId },
|
||||
data: {
|
||||
status: 'processing',
|
||||
@ -97,7 +124,7 @@ async function runVideoTranscodeTask(
|
||||
data: updateData,
|
||||
})
|
||||
|
||||
await (prisma as any).videoTranscodeJob.update({
|
||||
await videoTranscodeJob.update({
|
||||
where: { materialId },
|
||||
data: {
|
||||
status: 'done',
|
||||
@ -115,7 +142,7 @@ async function runVideoTranscodeTask(
|
||||
|
||||
console.error(`[Transcode] Persistent task failed for material ${materialId}:`, error)
|
||||
|
||||
await (prisma as any).videoTranscodeJob.update({
|
||||
await videoTranscodeJob.update({
|
||||
where: { materialId },
|
||||
data: {
|
||||
status: shouldRetry ? 'queued' : 'failed',
|
||||
@ -156,8 +183,9 @@ export async function enqueueVideoTranscodeJob(params: {
|
||||
}) {
|
||||
const { materialId, videoSourceUrl, needsCoverImage, forceRetry = false } = params
|
||||
const maxAttempts = getMaxAttempts()
|
||||
const videoTranscodeJob = requireVideoTranscodeJobDelegate()
|
||||
|
||||
await (prisma as any).videoTranscodeJob.upsert({
|
||||
await videoTranscodeJob.upsert({
|
||||
where: { materialId },
|
||||
create: {
|
||||
materialId,
|
||||
@ -195,8 +223,13 @@ export async function enqueueVideoTranscodeJob(params: {
|
||||
|
||||
export async function recoverStaleVideoTranscodeJobs() {
|
||||
const staleBefore = new Date(Date.now() - getLockTimeoutMs())
|
||||
const videoTranscodeJob = getVideoTranscodeJobDelegate()
|
||||
if (!videoTranscodeJob) {
|
||||
console.warn('[Transcode] 跳过启动恢复:Prisma Client 缺少 videoTranscodeJob,请执行数据库迁移并重新生成 Prisma Client')
|
||||
return { recovered: 0, hasMore: false, skipped: true }
|
||||
}
|
||||
|
||||
const staleJobs = await (prisma as any).videoTranscodeJob.findMany({
|
||||
const staleJobs = await videoTranscodeJob.findMany({
|
||||
where: {
|
||||
OR: [
|
||||
{ status: 'queued' },
|
||||
@ -243,7 +276,7 @@ export async function recoverStaleVideoTranscodeJobs() {
|
||||
const sourceUrl = job.material?.videoSourceUrl || job.videoSourceUrl
|
||||
if (!sourceUrl) continue
|
||||
|
||||
await (prisma as any).videoTranscodeJob.update({
|
||||
await videoTranscodeJob.update({
|
||||
where: { materialId: job.materialId },
|
||||
data: {
|
||||
status: 'queued',
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user