Spaces:

Bin29
/

ManimCat

Running

ManimCat / src /rendering /queue /processors /video.processor.ts

Sync from main: 5548f84 fix(workflow): preserve reference images across generation and recovery [skip ci]

b1df986 1 day ago

11.2 kB

/**
 * Video Processor
 * 任务处理器 - 主编排器
 */

import { performance } from 'node:perf_hooks'
import { videoQueue } from '../bull-config'
import { finalizeJobTimings, storeJobResult } from '../../../infrastructure/redis/job/redis-job-result-store'
import { deleteJobStage } from '../../../infrastructure/redis/job/redis-job-state-store'
import { touchJobTracking } from '../../../infrastructure/redis/job/redis-job-tracking-store'
import { clearJobCancelled } from '../../job/job-cancel-store'
import { JobCancelledError } from '../../../shared/errors/errors'
import { createLogger } from '../../../shared/logging/logger'
import type { VideoJobData } from '../../../types'
import { runEditFlow, runGenerationFlow, runPreGeneratedFlow } from './video-processor-flows'
import { getRetryMeta, shouldDisableQueueRetry, storeProcessingStage } from './video-processor-utils'
import { getCurrentJobLogSummary, getCurrentJobTokenUsage, runWithJobLogContext } from '../../job/job-log-context'
import { recordTokenUsage, type TokenUsageTotals } from '../../../access'
import { normalizeTimings } from '../../../shared/time/timings'

const logger = createLogger('VideoProcessor')
const activeExecutions = new Map<string, Promise<unknown>>()
const JOB_HEARTBEAT_MS = 15_000

const EMPTY_TOKEN_TOTALS: TokenUsageTotals = {
  promptTokens: 0,
  completionTokens: 0,
  totalTokens: 0,
  measuredCalls: 0,
  unmeasuredCalls: 0
}

/**
 * Writes this Job's cumulative usage to the CatKey ledger.
 *
 * Called only on a terminal outcome. Intermediate failed attempts are skipped on purpose: the
 * upsert is keyed on the Job id, so the attempt that finally succeeds would overwrite them anyway,
 * and a Job that never succeeds should not leave a half-counted row behind.
 *
 * The recorder itself decides whether anything is metered — an absent `catKeyId` means BYOK — so
 * there is no mode check here.
 */
async function recordJobTokenUsage(data: VideoJobData): Promise<void> {
  await recordTokenUsage({
    catKeyId: data.catKeyId,
    model: data.customApiConfig?.model ?? '',
    sourceKind: 'workflow_job',
    sourceId: data.jobId,
    totals: getCurrentJobTokenUsage()?.totals ?? EMPTY_TOKEN_TOTALS
  })
}

function emitJobSummary(args: {
  jobId: string
  taskType: 'pre-generated' | 'ai-edit' | 'generation'
  result: 'completed' | 'failed'
  outputMode: string
  timings?: Record<string, number>
  renderPeakMemoryMB?: number
  error?: string
  attempt?: number
  maxAttempts?: number
}): void {
  const tokenSummary = getCurrentJobLogSummary()
  logger.info('job_summary', {
    _logType: 'job_summary',
    jobId: args.jobId,
    taskType: args.taskType,
    result: args.result,
    outputMode: args.outputMode,
    attempt: args.attempt,
    maxAttempts: args.maxAttempts,
    timings: args.timings,
    renderPeakMemoryMB: args.renderPeakMemoryMB,
    error: args.error,
    tokens: tokenSummary
      ? {
          totals: tokenSummary.totals,
          calls: tokenSummary.calls
        }
      : {
          totals: {
            promptTokens: 0,
            completionTokens: 0,
            totalTokens: 0,
            measuredCalls: 0,
            unmeasuredCalls: 0
          },
          calls: []
        }
  })
}

async function executeVideoJob(job: any): Promise<unknown> {
  const executionStartMono = performance.now()
  const data = job.data as VideoJobData
  const contextAttempt = typeof job.attemptsMade === 'number' ? job.attemptsMade + 1 : 1

  return runWithJobLogContext(
    {
      jobId: data.jobId,
      outputMode: data.outputMode || 'video',
      attempts: contextAttempt
    },
    async () => {
  const {
    jobId,
    concept,
    quality,
    outputMode = 'video',
    preGeneratedCode,
    editCode,
    editInstructions,
    promptOverrides,
    referenceImages
  } = data

  logger.info('Processing video job', {
    jobId,
    concept,
    outputMode,
    quality,
    hasPreGeneratedCode: !!preGeneratedCode,
    hasEditRequest: !!editInstructions,
    referenceImageCount: referenceImages?.length || 0
  })

  const timings: Record<string, number> = {}
  const submittedAtMs = Date.parse(data.timestamp)
  if (Number.isFinite(submittedAtMs)) {
    const startedAtMs = typeof job.processedOn === 'number' ? job.processedOn : Date.now()
    timings.queue = Math.max(0, startedAtMs - submittedAtMs)
  }
  const retryMeta = getRetryMeta(job)
  const initialStage = preGeneratedCode
    ? 'rendering'
    : editCode && editInstructions
      ? 'generating'
      : 'analyzing'

  try {
    await storeProcessingStage(jobId, initialStage, {
      attempt: retryMeta.currentAttempt,
      status: 'processing',
    })

    let taskType: 'pre-generated' | 'ai-edit' | 'generation'
    let result
    if (preGeneratedCode) {
      result = await runPreGeneratedFlow({ job, data, promptOverrides, timings })
      taskType = 'pre-generated'
      logger.info('Job completed (pre-generated code)', { jobId, timings })
    } else if (editCode && editInstructions) {
      result = await runEditFlow({ job, data, promptOverrides, timings })
      taskType = 'ai-edit'
      logger.info('Job completed', { jobId, source: 'ai-edit', timings })
    } else {
      result = await runGenerationFlow({ job, data, promptOverrides, timings })
      taskType = 'generation'
      logger.info('Job completed', { jobId, source: 'generation', timings })
    }

    // Persist the spans that can only be known after the result write: store itself and the
    // monotonic, end-to-end total. `storeOutcome` is checked so a duplicate execution that
    // preserved an earlier completed result cannot rewrite that result's timing.
    timings.total = Math.max(0, (timings.queue || 0) + (performance.now() - executionStartMono))
    const finalizedAt = Date.now()
    if (result.storeOutcome === 'stored') {
      const finalized = await finalizeJobTimings(
        jobId,
        'completed',
        normalizeTimings(timings) ?? {},
        finalizedAt
      )
      if (!finalized) {
        logger.warn('Failed to persist final job timings', { jobId, status: 'completed' })
      }
    }

    // Terminal success: the authoritative `job:result:*` is now in place, so the intermediate
    // `job:stage:*` / `job:tracking:*` markers are no longer needed. Removing them keeps Redis
    // from showing a stale "rendering"/"generating" state after completion and prevents any future
    // stage-based consumer (e.g. an SSE progress feed) from reporting a finished job as in-flight.
    await deleteJobStage(jobId)

    emitJobSummary({
      jobId,
      taskType,
      result: 'completed',
      outputMode,
      timings,
      renderPeakMemoryMB: result.renderPeakMemoryMB,
      attempt: retryMeta.currentAttempt,
      maxAttempts: retryMeta.maxAttempts
    })
    await recordJobTokenUsage(data)
    return result
  } catch (error) {
    const errorMessage = error instanceof Error ? error.message : String(error)
    const cancelReason = error instanceof JobCancelledError ? error.details : undefined
    const currentRetryMeta = getRetryMeta(job)
    const disableQueueRetry = shouldDisableQueueRetry(errorMessage)
    const willQueueRetry = !disableQueueRetry && currentRetryMeta.hasRemainingAttempts

    if (disableQueueRetry) {
      try {
        job.discard()
        logger.warn('Queue retry disabled for exhausted code retry', {
          jobId,
          error: errorMessage,
          currentAttempt: currentRetryMeta.currentAttempt,
          maxAttempts: currentRetryMeta.maxAttempts
        })
      } catch (discardError) {
        logger.warn('Failed to discard job retry', { jobId, error: discardError })
      }
    }

    if (willQueueRetry) {
      logger.warn('Job attempt failed, Bull will retry', {
        jobId,
        error: errorMessage,
        currentAttempt: currentRetryMeta.currentAttempt,
        maxAttempts: currentRetryMeta.maxAttempts
      })
      throw error
    }

    logger.error('Job failed', {
      jobId,
      error: errorMessage,
      timings,
      currentAttempt: currentRetryMeta.currentAttempt,
      maxAttempts: currentRetryMeta.maxAttempts
    })

    const failureStoreStart = performance.now()
    const storeOutcome = await storeJobResult(jobId, {
      status: 'failed',
      data: {
        error: errorMessage,
        cancelReason,
        outputMode,
        timings: normalizeTimings(timings),
        tokenUsage: getCurrentJobTokenUsage()
      }
    })
    timings.store = performance.now() - failureStoreStart
    timings.total = Math.max(0, (timings.queue || 0) + (performance.now() - executionStartMono))
    if (storeOutcome === 'stored') {
      const finalized = await finalizeJobTimings(
        jobId,
        'failed',
        normalizeTimings(timings) ?? {},
        Date.now()
      )
      if (!finalized) {
        logger.warn('Failed to persist final job timings', { jobId, status: 'failed' })
      }
    }
    if (!cancelReason) {
      await clearJobCancelled(jobId)
    }

    // Terminal failure (no remaining Bull retries): same cleanup as success. The `willQueueRetry`
    // branch above threw before reaching here, so this only runs when the job is truly done.
    await deleteJobStage(jobId)

    // `preserved-completed` is the success-path guard; `preserved-failed` is the failure-path
    // guard used when an earlier execution already stored a completed result.
    if (storeOutcome === 'preserved-completed' || storeOutcome === 'preserved-failed') {
      logger.warn('Duplicate execution failed after a completed result; preserving success', {
        jobId,
        error: errorMessage,
        currentAttempt: currentRetryMeta.currentAttempt,
      })
      return {
        success: true,
        source: 'preserved-completed-result',
      }
    }

    emitJobSummary({
      jobId,
      taskType: editCode && editInstructions ? 'ai-edit' : preGeneratedCode ? 'pre-generated' : 'generation',
      result: 'failed',
      outputMode,
      timings,
      error: errorMessage,
      attempt: currentRetryMeta.currentAttempt,
      maxAttempts: currentRetryMeta.maxAttempts
    })

    // Terminal failure: the attempt consumed tokens it will never be credited for by a retry, so
    // the row is written here. `preserved-completed` returned above and must not reach this point.
    await recordJobTokenUsage(data)

    throw error
  }
    }
  )
}

videoQueue.process(async (job) => {
  const data = job.data as VideoJobData
  const jobId = data.jobId || String(job.id)
  const activeExecution = activeExecutions.get(jobId)

  if (activeExecution) {
    logger.warn('Duplicate Bull execution joined the existing in-process job', {
      jobId,
      attemptsMade: job.attemptsMade,
    })
    return activeExecution
  }

  const execution = executeVideoJob(job)
  activeExecutions.set(jobId, execution)
  const heartbeat = setInterval(() => {
    void touchJobTracking(jobId, (job.attemptsMade ?? 0) + 1)
  }, JOB_HEARTBEAT_MS)
  heartbeat.unref()

  try {
    return await execution
  } finally {
    clearInterval(heartbeat)
    if (activeExecutions.get(jobId) === execution) {
      activeExecutions.delete(jobId)
    }
  }
})

	/**
	* Video Processor
	* 任务处理器 - 主编排器
	*/

	import { performance } from 'node:perf_hooks'
	import { videoQueue } from '../bull-config'
	import { finalizeJobTimings, storeJobResult } from '../../../infrastructure/redis/job/redis-job-result-store'
	import { deleteJobStage } from '../../../infrastructure/redis/job/redis-job-state-store'
	import { touchJobTracking } from '../../../infrastructure/redis/job/redis-job-tracking-store'
	import { clearJobCancelled } from '../../job/job-cancel-store'
	import { JobCancelledError } from '../../../shared/errors/errors'
	import { createLogger } from '../../../shared/logging/logger'
	import type { VideoJobData } from '../../../types'
	import { runEditFlow, runGenerationFlow, runPreGeneratedFlow } from './video-processor-flows'
	import { getRetryMeta, shouldDisableQueueRetry, storeProcessingStage } from './video-processor-utils'
	import { getCurrentJobLogSummary, getCurrentJobTokenUsage, runWithJobLogContext } from '../../job/job-log-context'
	import { recordTokenUsage, type TokenUsageTotals } from '../../../access'
	import { normalizeTimings } from '../../../shared/time/timings'

	const logger = createLogger('VideoProcessor')
	const activeExecutions = new Map<string, Promise<unknown>>()
	const JOB_HEARTBEAT_MS = 15_000

	const EMPTY_TOKEN_TOTALS: TokenUsageTotals = {
	promptTokens: 0,
	completionTokens: 0,
	totalTokens: 0,
	measuredCalls: 0,
	unmeasuredCalls: 0
	}

	/**
	* Writes this Job's cumulative usage to the CatKey ledger.
	*
	* Called only on a terminal outcome. Intermediate failed attempts are skipped on purpose: the
	* upsert is keyed on the Job id, so the attempt that finally succeeds would overwrite them anyway,
	* and a Job that never succeeds should not leave a half-counted row behind.
	*
	* The recorder itself decides whether anything is metered — an absent `catKeyId` means BYOK — so
	* there is no mode check here.
	*/
	async function recordJobTokenUsage(data: VideoJobData): Promise<void> {
	await recordTokenUsage({
	catKeyId: data.catKeyId,
	model: data.customApiConfig?.model ?? '',
	sourceKind: 'workflow_job',
	sourceId: data.jobId,
	totals: getCurrentJobTokenUsage()?.totals ?? EMPTY_TOKEN_TOTALS
	})
	}

	function emitJobSummary(args: {
	jobId: string
	taskType: 'pre-generated' \| 'ai-edit' \| 'generation'
	result: 'completed' \| 'failed'
	outputMode: string
	timings?: Record<string, number>
	renderPeakMemoryMB?: number
	error?: string
	attempt?: number
	maxAttempts?: number
	}): void {
	const tokenSummary = getCurrentJobLogSummary()
	logger.info('job_summary', {
	_logType: 'job_summary',
	jobId: args.jobId,
	taskType: args.taskType,
	result: args.result,
	outputMode: args.outputMode,
	attempt: args.attempt,
	maxAttempts: args.maxAttempts,
	timings: args.timings,
	renderPeakMemoryMB: args.renderPeakMemoryMB,
	error: args.error,
	tokens: tokenSummary
	? {
	totals: tokenSummary.totals,
	calls: tokenSummary.calls
	}
	: {
	totals: {
	promptTokens: 0,
	completionTokens: 0,
	totalTokens: 0,
	measuredCalls: 0,
	unmeasuredCalls: 0
	},
	calls: []
	}
	})
	}

	async function executeVideoJob(job: any): Promise<unknown> {
	const executionStartMono = performance.now()
	const data = job.data as VideoJobData
	const contextAttempt = typeof job.attemptsMade === 'number' ? job.attemptsMade + 1 : 1

	return runWithJobLogContext(
	{
	jobId: data.jobId,
	outputMode: data.outputMode \|\| 'video',
	attempts: contextAttempt
	},
	async () => {
	const {
	jobId,
	concept,
	quality,
	outputMode = 'video',
	preGeneratedCode,
	editCode,
	editInstructions,
	promptOverrides,
	referenceImages
	} = data

	logger.info('Processing video job', {
	jobId,
	concept,
	outputMode,
	quality,
	hasPreGeneratedCode: !!preGeneratedCode,
	hasEditRequest: !!editInstructions,
	referenceImageCount: referenceImages?.length \|\| 0
	})

	const timings: Record<string, number> = {}
	const submittedAtMs = Date.parse(data.timestamp)
	if (Number.isFinite(submittedAtMs)) {
	const startedAtMs = typeof job.processedOn === 'number' ? job.processedOn : Date.now()
	timings.queue = Math.max(0, startedAtMs - submittedAtMs)
	}
	const retryMeta = getRetryMeta(job)
	const initialStage = preGeneratedCode
	? 'rendering'
	: editCode && editInstructions
	? 'generating'
	: 'analyzing'

	try {
	await storeProcessingStage(jobId, initialStage, {
	attempt: retryMeta.currentAttempt,
	status: 'processing',
	})

	let taskType: 'pre-generated' \| 'ai-edit' \| 'generation'
	let result
	if (preGeneratedCode) {
	result = await runPreGeneratedFlow({ job, data, promptOverrides, timings })
	taskType = 'pre-generated'
	logger.info('Job completed (pre-generated code)', { jobId, timings })
	} else if (editCode && editInstructions) {
	result = await runEditFlow({ job, data, promptOverrides, timings })
	taskType = 'ai-edit'
	logger.info('Job completed', { jobId, source: 'ai-edit', timings })
	} else {
	result = await runGenerationFlow({ job, data, promptOverrides, timings })
	taskType = 'generation'
	logger.info('Job completed', { jobId, source: 'generation', timings })
	}

	// Persist the spans that can only be known after the result write: store itself and the
	// monotonic, end-to-end total. `storeOutcome` is checked so a duplicate execution that
	// preserved an earlier completed result cannot rewrite that result's timing.
	timings.total = Math.max(0, (timings.queue \|\| 0) + (performance.now() - executionStartMono))
	const finalizedAt = Date.now()
	if (result.storeOutcome === 'stored') {
	const finalized = await finalizeJobTimings(
	jobId,
	'completed',
	normalizeTimings(timings) ?? {},
	finalizedAt
	)
	if (!finalized) {
	logger.warn('Failed to persist final job timings', { jobId, status: 'completed' })
	}
	}

	// Terminal success: the authoritative `job:result:*` is now in place, so the intermediate
	// `job:stage:` / `job:tracking:` markers are no longer needed. Removing them keeps Redis
	// from showing a stale "rendering"/"generating" state after completion and prevents any future
	// stage-based consumer (e.g. an SSE progress feed) from reporting a finished job as in-flight.
	await deleteJobStage(jobId)

	emitJobSummary({
	jobId,
	taskType,
	result: 'completed',
	outputMode,
	timings,
	renderPeakMemoryMB: result.renderPeakMemoryMB,
	attempt: retryMeta.currentAttempt,
	maxAttempts: retryMeta.maxAttempts
	})
	await recordJobTokenUsage(data)
	return result
	} catch (error) {
	const errorMessage = error instanceof Error ? error.message : String(error)
	const cancelReason = error instanceof JobCancelledError ? error.details : undefined
	const currentRetryMeta = getRetryMeta(job)
	const disableQueueRetry = shouldDisableQueueRetry(errorMessage)
	const willQueueRetry = !disableQueueRetry && currentRetryMeta.hasRemainingAttempts

	if (disableQueueRetry) {
	try {
	job.discard()
	logger.warn('Queue retry disabled for exhausted code retry', {
	jobId,
	error: errorMessage,
	currentAttempt: currentRetryMeta.currentAttempt,
	maxAttempts: currentRetryMeta.maxAttempts
	})
	} catch (discardError) {
	logger.warn('Failed to discard job retry', { jobId, error: discardError })
	}
	}

	if (willQueueRetry) {
	logger.warn('Job attempt failed, Bull will retry', {
	jobId,
	error: errorMessage,
	currentAttempt: currentRetryMeta.currentAttempt,
	maxAttempts: currentRetryMeta.maxAttempts
	})
	throw error
	}

	logger.error('Job failed', {
	jobId,
	error: errorMessage,
	timings,
	currentAttempt: currentRetryMeta.currentAttempt,
	maxAttempts: currentRetryMeta.maxAttempts
	})

	const failureStoreStart = performance.now()
	const storeOutcome = await storeJobResult(jobId, {
	status: 'failed',
	data: {
	error: errorMessage,
	cancelReason,
	outputMode,
	timings: normalizeTimings(timings),
	tokenUsage: getCurrentJobTokenUsage()
	}
	})
	timings.store = performance.now() - failureStoreStart
	timings.total = Math.max(0, (timings.queue \|\| 0) + (performance.now() - executionStartMono))
	if (storeOutcome === 'stored') {
	const finalized = await finalizeJobTimings(
	jobId,
	'failed',
	normalizeTimings(timings) ?? {},
	Date.now()
	)
	if (!finalized) {
	logger.warn('Failed to persist final job timings', { jobId, status: 'failed' })
	}
	}
	if (!cancelReason) {
	await clearJobCancelled(jobId)
	}

	// Terminal failure (no remaining Bull retries): same cleanup as success. The `willQueueRetry`
	// branch above threw before reaching here, so this only runs when the job is truly done.
	await deleteJobStage(jobId)

	// `preserved-completed` is the success-path guard; `preserved-failed` is the failure-path
	// guard used when an earlier execution already stored a completed result.
	if (storeOutcome === 'preserved-completed' \|\| storeOutcome === 'preserved-failed') {
	logger.warn('Duplicate execution failed after a completed result; preserving success', {
	jobId,
	error: errorMessage,
	currentAttempt: currentRetryMeta.currentAttempt,
	})
	return {
	success: true,
	source: 'preserved-completed-result',
	}
	}

	emitJobSummary({
	jobId,
	taskType: editCode && editInstructions ? 'ai-edit' : preGeneratedCode ? 'pre-generated' : 'generation',
	result: 'failed',
	outputMode,
	timings,
	error: errorMessage,
	attempt: currentRetryMeta.currentAttempt,
	maxAttempts: currentRetryMeta.maxAttempts
	})

	// Terminal failure: the attempt consumed tokens it will never be credited for by a retry, so
	// the row is written here. `preserved-completed` returned above and must not reach this point.
	await recordJobTokenUsage(data)

	throw error
	}
	}
	)
	}

	videoQueue.process(async (job) => {
	const data = job.data as VideoJobData
	const jobId = data.jobId \|\| String(job.id)
	const activeExecution = activeExecutions.get(jobId)

	if (activeExecution) {
	logger.warn('Duplicate Bull execution joined the existing in-process job', {
	jobId,
	attemptsMade: job.attemptsMade,
	})
	return activeExecution
	}

	const execution = executeVideoJob(job)
	activeExecutions.set(jobId, execution)
	const heartbeat = setInterval(() => {
	void touchJobTracking(jobId, (job.attemptsMade ?? 0) + 1)
	}, JOB_HEARTBEAT_MS)
	heartbeat.unref()

	try {
	return await execution
	} finally {
	clearInterval(heartbeat)
	if (activeExecutions.get(jobId) === execution) {
	activeExecutions.delete(jobId)
	}
	}
	})