7326e273e1
Protect upload cleanup and atomic completion paths, add durable media worker leases and brand asset validation, and introduce verified migration/bootstrap/deployment gates.
817 lines
28 KiB
TypeScript
817 lines
28 KiB
TypeScript
/**
|
||
* 视频转码持久化队列。
|
||
*
|
||
* 内存队列只负责降低当前进程的调度开销，数据库中的阶段记录和素材租约才是事实来源。
|
||
* 每次执行前必须先取得全局唯一租约，因此蓝绿容器并存时也只会有一个 ffmpeg 任务。
|
||
*/
|
||
|
||
import { randomUUID } from 'crypto'
|
||
import { access, mkdir, rm, writeFile } from 'fs/promises'
|
||
import path from 'path'
|
||
import { LEGACY_COMPATIBLE_VIDEO_MATERIAL_TYPES } from './video-material-types'
|
||
|
||
export const VIDEO_MATERIAL_TYPES = [...LEGACY_COMPATIBLE_VIDEO_MATERIAL_TYPES]
|
||
|
||
export const VIDEO_TRANSCODE_DRAIN_PROTOCOL = 'VIDEO_TRANSCODE_DRAIN_PROTOCOL_V1'
|
||
export const DEFAULT_TRANSCODE_LEASE_MS = 5 * 60 * 1000
|
||
export const DEFAULT_TRANSCODE_HEARTBEAT_MS = 30 * 1000
|
||
export const LEGACY_TRANSCODE_STALE_MS = 10 * 60 * 1000
|
||
|
||
const SAFE_MAX_CONCURRENT_IN_SHARED_CONTAINER = 1
|
||
const GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY = 193582731
|
||
// 不直接 SELECT 返回 void 的 pg_advisory_xact_lock；Prisma 无法反序列化 void 列。
|
||
const ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL = 'SELECT 1 AS locked FROM pg_advisory_xact_lock($1)'
|
||
// 必须在取得 advisory lock 后读取数据库当前时钟，不能使用进入事务前的应用时间。
|
||
const READ_DATABASE_CLOCK_SQL = 'SELECT clock_timestamp() AS "now"'
|
||
|
||
export type TranscodeStage = 'preview' | 'clip' | 'cover'
|
||
|
||
export type TranscodeTaskFn = (
|
||
materialId: string,
|
||
videoSourceUrl: string,
|
||
needsCoverImage: boolean,
|
||
leaseToken: string,
|
||
abortSignal: AbortSignal,
|
||
) => Promise<void>
|
||
|
||
interface QueuedTask {
|
||
materialId: string
|
||
videoSourceUrl: string
|
||
needsCoverImage: boolean
|
||
taskFn: TranscodeTaskFn
|
||
}
|
||
|
||
interface RecoverInterruptedTasksOptions {
|
||
requeuePending?: boolean
|
||
limit?: number
|
||
videoTypes?: string[]
|
||
}
|
||
|
||
interface VideoTranscodeQueueOptions {
|
||
maxConcurrent?: number
|
||
prisma?: any
|
||
workerId?: string
|
||
instanceId?: string
|
||
runtimeRoot?: string
|
||
leaseDurationMs?: number
|
||
heartbeatIntervalMs?: number
|
||
drainAccess?: typeof access
|
||
}
|
||
|
||
type LeaseClaim =
|
||
| { status: 'acquired'; token: string }
|
||
| { status: 'busy' }
|
||
| { status: 'skip' }
|
||
|
||
function safeWorkerId(value: string): string {
|
||
return value.replace(/[^A-Za-z0-9_.-]/g, '_').slice(0, 120) || `worker-${process.pid}`
|
||
}
|
||
|
||
function uniqueTasks(tasks: Array<{ materialId: string; videoSourceUrl: string; needsCoverImage: boolean }>) {
|
||
const seen = new Set<string>()
|
||
return tasks.filter((task) => {
|
||
if (!task.materialId || !task.videoSourceUrl || seen.has(task.materialId)) {
|
||
return false
|
||
}
|
||
seen.add(task.materialId)
|
||
return true
|
||
})
|
||
}
|
||
|
||
async function readDatabaseClock(client: any): Promise<Date> {
|
||
const rows = await client.$queryRawUnsafe(READ_DATABASE_CLOCK_SQL)
|
||
const value = rows?.[0]?.now
|
||
const now = value instanceof Date ? value : new Date(value)
|
||
if (!value || Number.isNaN(now.getTime())) {
|
||
throw new Error('无法读取数据库当前时间，拒绝执行视频转码租约操作')
|
||
}
|
||
return now
|
||
}
|
||
|
||
export class VideoTranscodeQueue {
|
||
private queue: QueuedTask[] = []
|
||
private processing = new Set<string>()
|
||
private progress = new Map<string, number>()
|
||
private stages = new Map<string, TranscodeStage>()
|
||
private prisma: any
|
||
private readonly maxConcurrent: number
|
||
private readonly workerId: string
|
||
private readonly instanceId: string
|
||
private readonly runtimeRoot: string
|
||
private readonly leaseDurationMs: number
|
||
private readonly heartbeatIntervalMs: number
|
||
private readonly drainAccess: typeof access
|
||
private pumping = false
|
||
private recoveryPromise: Promise<number> | null = null
|
||
private retryTimer: ReturnType<typeof setTimeout> | null = null
|
||
|
||
constructor(options: VideoTranscodeQueueOptions | number = {}) {
|
||
const normalizedOptions = typeof options === 'number' ? { maxConcurrent: options } : options
|
||
const envConcurrent = Number.parseInt(process.env.VIDEO_TRANSCODE_MAX_CONCURRENT || '', 10)
|
||
const requested = Number.isFinite(envConcurrent)
|
||
? envConcurrent
|
||
: normalizedOptions.maxConcurrent ?? SAFE_MAX_CONCURRENT_IN_SHARED_CONTAINER
|
||
|
||
this.maxConcurrent = Math.max(1, Math.min(SAFE_MAX_CONCURRENT_IN_SHARED_CONTAINER, requested))
|
||
this.prisma = normalizedOptions.prisma || null
|
||
this.workerId = safeWorkerId(
|
||
normalizedOptions.workerId || process.env.VIDEO_TRANSCODE_WORKER_ID || process.env.HOSTNAME || `worker-${process.pid}`,
|
||
)
|
||
this.instanceId = safeWorkerId(
|
||
normalizedOptions.instanceId
|
||
|| process.env.VIDEO_TRANSCODE_INSTANCE_ID
|
||
|| `${this.workerId}-${process.pid}`,
|
||
)
|
||
this.runtimeRoot = normalizedOptions.runtimeRoot
|
||
|| process.env.VIDEO_TRANSCODE_RUNTIME_DIR
|
||
|| path.join(process.env.UPLOAD_DIR || path.join(process.cwd(), 'uploads'), '.video-transcode-runtime')
|
||
this.leaseDurationMs = normalizedOptions.leaseDurationMs || DEFAULT_TRANSCODE_LEASE_MS
|
||
this.heartbeatIntervalMs = normalizedOptions.heartbeatIntervalMs || DEFAULT_TRANSCODE_HEARTBEAT_MS
|
||
this.drainAccess = normalizedOptions.drainAccess || access
|
||
|
||
console.log(`[TranscodeQueue] Initialized worker=${this.workerId}, maxConcurrent=${this.maxConcurrent}`)
|
||
}
|
||
|
||
private async getPrisma() {
|
||
if (!this.prisma) {
|
||
const { prisma } = await import('@/lib/prisma')
|
||
this.prisma = prisma
|
||
}
|
||
return this.prisma
|
||
}
|
||
|
||
private capabilityPath() {
|
||
return path.join(this.runtimeRoot, 'capabilities', `${this.workerId}.v1`)
|
||
}
|
||
|
||
private drainPath() {
|
||
return path.join(this.runtimeRoot, 'drain', this.workerId)
|
||
}
|
||
|
||
private activePath() {
|
||
return path.join(this.runtimeRoot, 'active', `${this.workerId}.json`)
|
||
}
|
||
|
||
async ensureDrainProtocolCapability(): Promise<void> {
|
||
const target = this.capabilityPath()
|
||
await mkdir(path.dirname(target), { recursive: true })
|
||
await writeFile(
|
||
target,
|
||
`${VIDEO_TRANSCODE_DRAIN_PROTOCOL}\nworkerId=${this.workerId}\ninstanceId=${this.instanceId}\n`,
|
||
{ encoding: 'utf8', mode: 0o600 },
|
||
)
|
||
}
|
||
|
||
private async isDraining(): Promise<boolean> {
|
||
try {
|
||
await this.drainAccess(this.drainPath())
|
||
return true
|
||
} catch (error) {
|
||
const code = (error as NodeJS.ErrnoException | undefined)?.code
|
||
if (code === 'ENOENT') {
|
||
return false
|
||
}
|
||
|
||
// NAS/runtime 状态无法读取时必须 fail-closed。把 EACCES/EIO/ESTALE
|
||
// 等错误视为仍在 drain，pump 会保留任务并在下一轮重试，绝不启动 FFmpeg。
|
||
console.error(`[TranscodeQueue] Cannot verify drain marker for ${this.workerId}; treating worker as draining:`, error)
|
||
return true
|
||
}
|
||
}
|
||
|
||
private async writeActiveMarker(materialId: string, leaseToken: string): Promise<void> {
|
||
const target = this.activePath()
|
||
await mkdir(path.dirname(target), { recursive: true })
|
||
await writeFile(target, JSON.stringify({
|
||
protocol: VIDEO_TRANSCODE_DRAIN_PROTOCOL,
|
||
workerId: this.workerId,
|
||
instanceId: this.instanceId,
|
||
materialId,
|
||
leaseToken,
|
||
pid: process.pid,
|
||
heartbeatAt: new Date().toISOString(),
|
||
}), { encoding: 'utf8', mode: 0o600 })
|
||
}
|
||
|
||
private async removeActiveMarker(): Promise<void> {
|
||
await rm(this.activePath(), { force: true }).catch((error) => {
|
||
console.error(`[TranscodeQueue] Failed to remove active marker for ${this.workerId}:`, error)
|
||
})
|
||
}
|
||
|
||
private async saveStageProgress(
|
||
materialId: string,
|
||
stage: TranscodeStage,
|
||
status: 'waiting' | 'processing' | 'completed' | 'failed',
|
||
progress = 0,
|
||
errorMessage: string | null = null,
|
||
leaseToken?: string,
|
||
): Promise<boolean> {
|
||
const prisma = await this.getPrisma()
|
||
const writeProgress = async (client: any) => {
|
||
await client.videoTranscodeProgress.upsert({
|
||
where: { materialId_stage: { materialId, stage } },
|
||
create: {
|
||
materialId,
|
||
stage,
|
||
status,
|
||
progress,
|
||
errorMessage,
|
||
lastHeartbeat: new Date(),
|
||
},
|
||
update: {
|
||
status,
|
||
progress,
|
||
errorMessage,
|
||
lastHeartbeat: status === 'processing' ? new Date() : undefined,
|
||
completedAt: status === 'completed' || status === 'failed' ? new Date() : undefined,
|
||
},
|
||
})
|
||
}
|
||
|
||
if (!leaseToken) {
|
||
await writeProgress(prisma)
|
||
return true
|
||
}
|
||
|
||
// 进度表本身没有租约列，因此必须和 claim/release 共用同一把事务锁，
|
||
// 在锁内校验 token 后再写入，避免旧进程在校验与 upsert 之间越过新租约。
|
||
return prisma.$transaction(async (tx: any) => {
|
||
await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
const current = await tx.material.count({
|
||
where: {
|
||
id: materialId,
|
||
transcodeLeaseToken: leaseToken,
|
||
transcodeLeaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
})
|
||
if (current !== 1) {
|
||
return false
|
||
}
|
||
await writeProgress(tx)
|
||
return true
|
||
})
|
||
}
|
||
|
||
private async persistWaitingStages(task: Pick<QueuedTask, 'materialId' | 'needsCoverImage'>): Promise<void> {
|
||
const stages: TranscodeStage[] = ['preview', 'clip']
|
||
if (task.needsCoverImage) {
|
||
stages.push('cover')
|
||
}
|
||
for (const stage of stages) {
|
||
await this.saveStageProgress(task.materialId, stage, 'waiting')
|
||
}
|
||
}
|
||
|
||
private async prepareMaterialForQueue(materialId: string): Promise<boolean> {
|
||
const prisma = await this.getPrisma()
|
||
const prepared = await prisma.$transaction(async (tx: any) => {
|
||
await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
return tx.material.updateMany({
|
||
where: {
|
||
id: materialId,
|
||
deletedAt: null,
|
||
videoSourceUrl: { not: null },
|
||
OR: [
|
||
{ transcodeLeaseToken: null },
|
||
{ transcodeLeaseExpiresAt: null },
|
||
{ transcodeLeaseExpiresAt: { lte: databaseNow } },
|
||
],
|
||
},
|
||
data: { videoProcessingStatus: 'pending' },
|
||
})
|
||
})
|
||
return prepared.count === 1
|
||
}
|
||
|
||
private enqueueLocal(task: QueuedTask): boolean {
|
||
if (this.processing.has(task.materialId) || this.queue.some((queued) => queued.materialId === task.materialId)) {
|
||
return false
|
||
}
|
||
this.queue.push(task)
|
||
return true
|
||
}
|
||
|
||
/** 持久化等待状态后立即返回，不等待 ffmpeg 完成。 */
|
||
async add(
|
||
materialId: string,
|
||
videoSourceUrl: string,
|
||
needsCoverImage: boolean,
|
||
taskFn: TranscodeTaskFn,
|
||
): Promise<void> {
|
||
const task = { materialId, videoSourceUrl, needsCoverImage, taskFn }
|
||
if (this.processing.has(materialId) || this.queue.some((queued) => queued.materialId === materialId)) {
|
||
return
|
||
}
|
||
|
||
if (!(await this.prepareMaterialForQueue(materialId))) {
|
||
return
|
||
}
|
||
await this.persistWaitingStages(task)
|
||
if (this.enqueueLocal(task)) {
|
||
void this.pump()
|
||
}
|
||
}
|
||
|
||
private schedulePump(delayMs = 2_000): void {
|
||
if (this.retryTimer) {
|
||
return
|
||
}
|
||
this.retryTimer = setTimeout(() => {
|
||
this.retryTimer = null
|
||
void this.pump()
|
||
}, delayMs)
|
||
this.retryTimer.unref?.()
|
||
}
|
||
|
||
private async tryAcquireGlobalLease(materialId: string): Promise<LeaseClaim> {
|
||
const prisma = await this.getPrisma()
|
||
const token = randomUUID()
|
||
|
||
return prisma.$transaction(async (tx: any) => {
|
||
await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
const expiresAt = new Date(databaseNow.getTime() + this.leaseDurationMs)
|
||
|
||
const activeLease = await tx.material.findFirst({
|
||
where: {
|
||
transcodeLeaseToken: { not: null },
|
||
transcodeLeaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
select: { id: true },
|
||
})
|
||
if (activeLease) {
|
||
return { status: 'busy' } as const
|
||
}
|
||
|
||
const material = await tx.material.findUnique({
|
||
where: { id: materialId },
|
||
select: {
|
||
id: true,
|
||
deletedAt: true,
|
||
videoSourceUrl: true,
|
||
videoProcessingStatus: true,
|
||
},
|
||
})
|
||
if (!material || material.deletedAt || !material.videoSourceUrl || material.videoProcessingStatus === 'done') {
|
||
return { status: 'skip' } as const
|
||
}
|
||
|
||
const claimed = await tx.material.updateMany({
|
||
where: {
|
||
id: materialId,
|
||
OR: [
|
||
{ transcodeLeaseToken: null },
|
||
{ transcodeLeaseExpiresAt: null },
|
||
{ transcodeLeaseExpiresAt: { lte: databaseNow } },
|
||
],
|
||
},
|
||
data: {
|
||
transcodeLeaseToken: token,
|
||
transcodeLeaseExpiresAt: expiresAt,
|
||
videoProcessingStatus: 'processing',
|
||
transcodeLastError: null,
|
||
},
|
||
})
|
||
|
||
return claimed.count === 1
|
||
? { status: 'acquired', token } as const
|
||
: { status: 'busy' } as const
|
||
})
|
||
}
|
||
|
||
async isLeaseCurrent(materialId: string, leaseToken: string): Promise<boolean> {
|
||
const prisma = await this.getPrisma()
|
||
return prisma.$transaction(async (tx: any) => {
|
||
await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
const count = await tx.material.count({
|
||
where: {
|
||
id: materialId,
|
||
transcodeLeaseToken: leaseToken,
|
||
transcodeLeaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
})
|
||
return count === 1
|
||
})
|
||
}
|
||
|
||
async renewLease(materialId: string, leaseToken: string): Promise<boolean> {
|
||
const prisma = await this.getPrisma()
|
||
return prisma.$transaction(async (tx: any) => {
|
||
// renew/acquire/release/progress 必须共用同一把事务锁。这样旧 worker
|
||
// 的续租不可能在新 worker 已领取另一素材后再“复活”为第二个有效租约。
|
||
await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
const renewed = await tx.material.updateMany({
|
||
where: {
|
||
id: materialId,
|
||
transcodeLeaseToken: leaseToken,
|
||
transcodeLeaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
data: { transcodeLeaseExpiresAt: new Date(databaseNow.getTime() + this.leaseDurationMs) },
|
||
})
|
||
return renewed.count === 1
|
||
})
|
||
}
|
||
|
||
async releaseLease(materialId: string, leaseToken: string, resetToPending = false): Promise<boolean> {
|
||
const prisma = await this.getPrisma()
|
||
return prisma.$transaction(async (tx: any) => {
|
||
await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
|
||
const released = await tx.material.updateMany({
|
||
where: { id: materialId, transcodeLeaseToken: leaseToken },
|
||
data: {
|
||
transcodeLeaseToken: null,
|
||
transcodeLeaseExpiresAt: null,
|
||
...(resetToPending ? { videoProcessingStatus: 'pending' } : {}),
|
||
},
|
||
})
|
||
return released.count === 1
|
||
})
|
||
}
|
||
|
||
private async runClaimedTask(task: QueuedTask, leaseToken: string): Promise<void> {
|
||
this.processing.add(task.materialId)
|
||
let leaseLost = false
|
||
let heartbeatTimer: ReturnType<typeof setTimeout> | null = null
|
||
let heartbeatInFlight: Promise<void> = Promise.resolve()
|
||
let heartbeatStopped = false
|
||
let requeueAfterDrain = false
|
||
const abortController = new AbortController()
|
||
const scheduleHeartbeat = () => {
|
||
heartbeatTimer = setTimeout(() => {
|
||
heartbeatTimer = null
|
||
heartbeatInFlight = (async () => {
|
||
try {
|
||
const renewed = await this.renewLease(task.materialId, leaseToken)
|
||
if (!renewed) {
|
||
leaseLost = true
|
||
heartbeatStopped = true
|
||
abortController.abort(new Error('视频转码租约已失效'))
|
||
console.error(`[TranscodeQueue] Lease lost for ${task.materialId}; stale result writes will be rejected`)
|
||
return
|
||
}
|
||
await this.writeActiveMarker(task.materialId, leaseToken)
|
||
} catch (error) {
|
||
leaseLost = true
|
||
heartbeatStopped = true
|
||
abortController.abort(error instanceof Error ? error : new Error(String(error)))
|
||
console.error(`[TranscodeQueue] Lease heartbeat failed for ${task.materialId}:`, error)
|
||
}
|
||
})().finally(() => {
|
||
if (!heartbeatStopped) {
|
||
scheduleHeartbeat()
|
||
}
|
||
})
|
||
}, this.heartbeatIntervalMs)
|
||
;(heartbeatTimer as NodeJS.Timeout).unref?.()
|
||
}
|
||
|
||
try {
|
||
// active 标记写入成功后才允许启动 ffmpeg；标记失败时保持 fail-closed。
|
||
await this.writeActiveMarker(task.materialId, leaseToken)
|
||
// drain 可能恰好出现在 pump 二次检查与 active 标记写入之间。
|
||
// 写入 active 后再核对一次；命中时由 finally 释放租约、清理标记并重新入队，绝不启动 ffmpeg。
|
||
if (await this.isDraining()) {
|
||
requeueAfterDrain = true
|
||
console.log(`[TranscodeQueue] Drain detected after active marker for ${task.materialId}; task stays pending`)
|
||
return
|
||
}
|
||
scheduleHeartbeat()
|
||
|
||
await task.taskFn(
|
||
task.materialId,
|
||
task.videoSourceUrl,
|
||
task.needsCoverImage,
|
||
leaseToken,
|
||
abortController.signal,
|
||
)
|
||
if (leaseLost || !(await this.isLeaseCurrent(task.materialId, leaseToken))) {
|
||
console.warn(`[TranscodeQueue] Ignoring completion from stale worker for ${task.materialId}`)
|
||
} else {
|
||
console.log(`[TranscodeQueue] Task completed for ${task.materialId}`)
|
||
}
|
||
} catch (error) {
|
||
console.error(`[TranscodeQueue] Task failed for ${task.materialId}:`, error)
|
||
} finally {
|
||
heartbeatStopped = true
|
||
if (heartbeatTimer) {
|
||
clearTimeout(heartbeatTimer)
|
||
}
|
||
// 等待已经开始的续租/active 标记写入结束，再释放租约并删除标记，
|
||
// 防止异步心跳在 finally 之后重新制造一个陈旧 active 文件。
|
||
await heartbeatInFlight.catch(() => {})
|
||
this.progress.delete(task.materialId)
|
||
this.stages.delete(task.materialId)
|
||
// drain 或心跳/marker/租约中断都属于可恢复中断。只有当前 token 仍归本
|
||
// worker 时才会清租约并回 pending；若 token 已被新 worker 接管，CAS 无操作。
|
||
const resetInterruptedTask = requeueAfterDrain || leaseLost || abortController.signal.aborted
|
||
await this.releaseLease(task.materialId, leaseToken, resetInterruptedTask).catch((error) => {
|
||
console.error(`[TranscodeQueue] Lease release failed for ${task.materialId}:`, error)
|
||
})
|
||
await this.removeActiveMarker()
|
||
this.processing.delete(task.materialId)
|
||
if (requeueAfterDrain && !this.queue.some((queued) => queued.materialId === task.materialId)) {
|
||
this.queue.unshift(task)
|
||
}
|
||
void this.pump()
|
||
}
|
||
}
|
||
|
||
private async pump(): Promise<void> {
|
||
if (this.pumping || this.processing.size >= this.maxConcurrent || this.queue.length === 0) {
|
||
return
|
||
}
|
||
this.pumping = true
|
||
try {
|
||
await this.ensureDrainProtocolCapability()
|
||
if (await this.isDraining()) {
|
||
this.schedulePump(2_000)
|
||
return
|
||
}
|
||
|
||
const task = this.queue.shift()
|
||
if (!task) {
|
||
return
|
||
}
|
||
const claim = await this.tryAcquireGlobalLease(task.materialId)
|
||
if (claim.status === 'busy') {
|
||
this.queue.push(task)
|
||
this.schedulePump(2_000)
|
||
return
|
||
}
|
||
if (claim.status === 'skip') {
|
||
this.progress.delete(task.materialId)
|
||
this.stages.delete(task.materialId)
|
||
this.schedulePump(0)
|
||
return
|
||
}
|
||
|
||
// drain 标记可能在数据库 claim 进行期间出现；取得租约后再次检查，
|
||
// 确保部署脚本确认 idle 后不会再启动新的 ffmpeg。
|
||
if (await this.isDraining()) {
|
||
await this.releaseLease(task.materialId, claim.token, true)
|
||
this.queue.unshift(task)
|
||
this.schedulePump(2_000)
|
||
return
|
||
}
|
||
|
||
void this.runClaimedTask(task, claim.token)
|
||
} catch (error) {
|
||
console.error('[TranscodeQueue] Queue pump failed:', error)
|
||
this.schedulePump(2_000)
|
||
} finally {
|
||
this.pumping = false
|
||
}
|
||
}
|
||
|
||
async updateProgress(
|
||
materialId: string,
|
||
leaseToken: string,
|
||
progress: number,
|
||
stage?: TranscodeStage,
|
||
): Promise<boolean> {
|
||
const normalized = Math.min(100, Math.max(0, progress))
|
||
if (stage) {
|
||
const saved = await this.saveStageProgress(
|
||
materialId,
|
||
stage,
|
||
normalized >= 100 ? 'completed' : 'processing',
|
||
normalized,
|
||
null,
|
||
leaseToken,
|
||
)
|
||
if (!saved) {
|
||
return false
|
||
}
|
||
this.progress.set(materialId, normalized)
|
||
this.stages.set(materialId, stage)
|
||
return true
|
||
}
|
||
if (!(await this.isLeaseCurrent(materialId, leaseToken))) {
|
||
return false
|
||
}
|
||
this.progress.set(materialId, normalized)
|
||
return true
|
||
}
|
||
|
||
getStatus() {
|
||
return {
|
||
queueLength: this.queue.length,
|
||
processingCount: this.processing.size,
|
||
processingIds: Array.from(this.processing),
|
||
maxConcurrent: this.maxConcurrent,
|
||
progress: this.progress,
|
||
stages: this.stages,
|
||
waiting: this.queue.map((task) => task.materialId),
|
||
}
|
||
}
|
||
|
||
getDetailedStatus() {
|
||
return {
|
||
queue: this.queue.map((task) => task.materialId),
|
||
processing: Array.from(this.processing),
|
||
maxConcurrent: this.maxConcurrent,
|
||
}
|
||
}
|
||
|
||
async loadWaitingTasks(limit?: number): Promise<Array<{ materialId: string; videoSourceUrl: string; needsCoverImage: boolean }>> {
|
||
const prisma = await this.getPrisma()
|
||
const rows = await prisma.videoTranscodeProgress.findMany({
|
||
where: { status: { in: ['waiting', 'processing'] } },
|
||
include: {
|
||
material: {
|
||
select: {
|
||
videoSourceUrl: true,
|
||
videoCoverUrl: true,
|
||
videoProcessingStatus: true,
|
||
deletedAt: true,
|
||
},
|
||
},
|
||
},
|
||
orderBy: [{ priority: 'desc' }, { createdAt: 'asc' }],
|
||
...(limit && limit > 0 ? { take: limit * 3 } : {}),
|
||
})
|
||
|
||
const tasks = rows
|
||
.filter((row: any) => row.material
|
||
&& !row.material.deletedAt
|
||
&& row.material.videoSourceUrl
|
||
&& row.material.videoProcessingStatus !== 'done'
|
||
&& row.material.videoProcessingStatus !== 'failed')
|
||
.map((row: any) => ({
|
||
materialId: String(row.materialId),
|
||
videoSourceUrl: String(row.material.videoSourceUrl),
|
||
needsCoverImage: !row.material.videoCoverUrl,
|
||
}))
|
||
return uniqueTasks(tasks).slice(0, limit && limit > 0 ? limit : undefined)
|
||
}
|
||
|
||
private async recoverExpiredAndLegacyLeases(videoTypes: string[]): Promise<number> {
|
||
const prisma = await this.getPrisma()
|
||
// 预筛也必须使用数据库时钟。最终回收仍会在 advisory lock 内再次以
|
||
// 数据库时钟做 CAS，避免应用容器时钟偏快/偏慢造成漏恢复或误恢复。
|
||
const databaseNow = await readDatabaseClock(prisma)
|
||
const staleBefore = new Date(databaseNow.getTime() - LEGACY_TRANSCODE_STALE_MS)
|
||
let recovered = 0
|
||
|
||
const expired = await prisma.material.findMany({
|
||
where: {
|
||
materialType: { in: videoTypes },
|
||
videoProcessingStatus: 'processing',
|
||
videoSourceUrl: { not: null },
|
||
deletedAt: null,
|
||
transcodeLeaseToken: { not: null },
|
||
transcodeLeaseExpiresAt: { lte: databaseNow },
|
||
},
|
||
select: { id: true, transcodeLeaseToken: true, transcodeLeaseExpiresAt: true },
|
||
})
|
||
|
||
for (const material of expired) {
|
||
const released = await prisma.$transaction(async (tx: any) => {
|
||
await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
const result = await tx.material.updateMany({
|
||
where: {
|
||
id: material.id,
|
||
transcodeLeaseToken: material.transcodeLeaseToken,
|
||
transcodeLeaseExpiresAt: { lte: databaseNow },
|
||
},
|
||
data: {
|
||
videoProcessingStatus: 'pending',
|
||
transcodeLeaseToken: null,
|
||
transcodeLeaseExpiresAt: null,
|
||
},
|
||
})
|
||
return result.count
|
||
})
|
||
recovered += released
|
||
}
|
||
|
||
// 兼容首次上线前留下的旧任务：只有素材和阶段心跳都已静默 10 分钟才回收。
|
||
const legacy = await prisma.material.findMany({
|
||
where: {
|
||
materialType: { in: videoTypes },
|
||
videoProcessingStatus: 'processing',
|
||
videoSourceUrl: { not: null },
|
||
deletedAt: null,
|
||
transcodeLeaseToken: null,
|
||
updatedAt: { lt: staleBefore },
|
||
},
|
||
select: {
|
||
id: true,
|
||
updatedAt: true,
|
||
transcodeProgresses: {
|
||
select: { lastHeartbeat: true },
|
||
orderBy: { lastHeartbeat: 'desc' },
|
||
take: 1,
|
||
},
|
||
},
|
||
})
|
||
|
||
for (const material of legacy) {
|
||
const heartbeat = material.transcodeProgresses?.[0]?.lastHeartbeat
|
||
if (heartbeat && heartbeat >= staleBefore) {
|
||
continue
|
||
}
|
||
const released = await prisma.$transaction(async (tx: any) => {
|
||
await tx.$queryRawUnsafe(ACQUIRE_GLOBAL_TRANSCODE_LOCK_SQL, GLOBAL_TRANSCODE_ADVISORY_LOCK_KEY)
|
||
const result = await tx.material.updateMany({
|
||
where: {
|
||
id: material.id,
|
||
videoProcessingStatus: 'processing',
|
||
transcodeLeaseToken: null,
|
||
updatedAt: material.updatedAt,
|
||
},
|
||
data: { videoProcessingStatus: 'pending' },
|
||
})
|
||
return result.count
|
||
})
|
||
recovered += released
|
||
}
|
||
|
||
return recovered
|
||
}
|
||
|
||
private async loadPendingMaterials(videoTypes: string[], limit?: number) {
|
||
const prisma = await this.getPrisma()
|
||
const rows = await prisma.material.findMany({
|
||
where: {
|
||
materialType: { in: videoTypes },
|
||
videoProcessingStatus: 'pending',
|
||
videoSourceUrl: { not: null },
|
||
deletedAt: null,
|
||
},
|
||
orderBy: { createdAt: 'asc' },
|
||
select: { id: true, videoSourceUrl: true, videoCoverUrl: true },
|
||
...(limit && limit > 0 ? { take: limit } : {}),
|
||
})
|
||
return rows.map((row: any) => ({
|
||
materialId: String(row.id),
|
||
videoSourceUrl: String(row.videoSourceUrl),
|
||
needsCoverImage: !row.videoCoverUrl,
|
||
}))
|
||
}
|
||
|
||
/**
|
||
* 回收过期租约并重新装载持久化任务。恢复过程不删除任何 S3 对象或 canonical 源文件。
|
||
* 同一进程的重复扫描会复用同一个 Promise，避免 60 秒定时器重叠。
|
||
*/
|
||
async runRecoveryScan(
|
||
taskFn: TranscodeTaskFn,
|
||
options: RecoverInterruptedTasksOptions = {},
|
||
): Promise<number> {
|
||
if (this.recoveryPromise) {
|
||
return this.recoveryPromise
|
||
}
|
||
|
||
this.recoveryPromise = (async () => {
|
||
await this.ensureDrainProtocolCapability()
|
||
const videoTypes = options.videoTypes?.length ? options.videoTypes : VIDEO_MATERIAL_TYPES
|
||
const recoveredLeases = await this.recoverExpiredAndLegacyLeases(videoTypes)
|
||
const persisted = await this.loadWaitingTasks(options.limit)
|
||
const pending = options.requeuePending === false
|
||
? []
|
||
: await this.loadPendingMaterials(videoTypes, options.limit)
|
||
const tasks = uniqueTasks([...persisted, ...pending])
|
||
let enqueued = 0
|
||
|
||
for (const task of tasks) {
|
||
if (this.enqueueLocal({ ...task, taskFn })) {
|
||
enqueued += 1
|
||
}
|
||
}
|
||
if (enqueued > 0) {
|
||
void this.pump()
|
||
}
|
||
console.log(`[TranscodeQueue] Recovery scan: released=${recoveredLeases}, enqueued=${enqueued}`)
|
||
return enqueued
|
||
})()
|
||
|
||
try {
|
||
return await this.recoveryPromise
|
||
} finally {
|
||
this.recoveryPromise = null
|
||
}
|
||
}
|
||
|
||
async recoverOnStartup(taskFn: TranscodeTaskFn): Promise<number> {
|
||
return this.runRecoveryScan(taskFn, { requeuePending: true, videoTypes: VIDEO_MATERIAL_TYPES })
|
||
}
|
||
|
||
/** 兼容旧调用签名；cleanupFn 被有意忽略，恢复不得删除当前引用对象。 */
|
||
async recoverInterruptedTasks(
|
||
taskFn: TranscodeTaskFn,
|
||
_cleanupFn?: unknown,
|
||
options: RecoverInterruptedTasksOptions = {},
|
||
): Promise<{ recovered: number; requeued: number }> {
|
||
const requeued = await this.runRecoveryScan(taskFn, { ...options, requeuePending: true })
|
||
return { recovered: 0, requeued }
|
||
}
|
||
}
|
||
|
||
export const transcodeQueue = new VideoTranscodeQueue()
|