gongxiangp2/lib/task-detail-upload-runtime.ts

import { spawn, type ChildProcessWithoutNullStreams } from 'child_process'
import { randomUUID } from 'crypto'
import { access, lstat, mkdir, open, rename, rm } from 'fs/promises'
import path from 'path'

import { getUploadRootDir } from '@/lib/upload-paths'

export const TASK_DETAIL_UPLOAD_DRAIN_PROTOCOL = 'TASK_DETAIL_UPLOAD_DRAIN_PROTOCOL_V1'

const DEFAULT_EXECUTION_LOCK_TIMEOUT_MS = 5_000
const EXECUTION_LOCK_READY_MARKER = 'TASK_DETAIL_UPLOAD_EXECUTION_LOCK_READY\n'
const EXECUTION_LOCK_EXIT_TIMEOUT_MS = 5_000
const EXECUTION_LOCK_FILE_NAME = 'task-detail-upload-execution.lock'

export type TaskDetailUploadExecutionLockRelease = (() => Promise<void>) & {
  assertHeld: () => void
}

export class TaskDetailUploadExecutionLockUnavailableError extends Error {
  constructor(message = '任务详情处理器正忙，请稍后重试') {
    super(message)
    this.name = 'TaskDetailUploadExecutionLockUnavailableError'
  }
}

export class TaskDetailUploadExecutionLockLostError extends Error {
  constructor() {
    super('任务详情执行锁意外中断，本次结果已安全拒绝')
    this.name = 'TaskDetailUploadExecutionLockLostError'
  }
}

interface PendingExecutionLockRequest {
  resolve: (release: TaskDetailUploadExecutionLockRelease) => void
  reject: (error: Error) => void
  timer: ReturnType<typeof setTimeout>
}

/** Windows/Jest 下使用的进程内锁；生产 Linux 必须走 flock。 */
class InProcessTaskDetailUploadExecutionLock {
  private held = false
  private readonly waiting: PendingExecutionLockRequest[] = []

  acquire(timeoutMs: number): Promise<TaskDetailUploadExecutionLockRelease> {
    if (!this.held && this.waiting.length === 0) {
      return Promise.resolve(this.grant())
    }

    return new Promise<TaskDetailUploadExecutionLockRelease>((resolve, reject) => {
      const request: PendingExecutionLockRequest = {
        resolve,
        reject,
        timer: setTimeout(() => {
          const index = this.waiting.indexOf(request)
          if (index >= 0) this.waiting.splice(index, 1)
          reject(new TaskDetailUploadExecutionLockUnavailableError())
        }, timeoutMs),
      }
      request.timer.unref?.()
      this.waiting.push(request)
    })
  }

  private grant(): TaskDetailUploadExecutionLockRelease {
    this.held = true
    let released = false
    const release = async () => {
      if (released) return
      released = true
      const next = this.waiting.shift()
      if (next) {
        clearTimeout(next.timer)
        next.resolve(this.grant())
      } else {
        this.held = false
      }
    }
    release.assertHeld = () => {
      if (released || !this.held) throw new TaskDetailUploadExecutionLockLostError()
    }
    return release
  }
}

const inProcessExecutionLock = new InProcessTaskDetailUploadExecutionLock()

function safeRuntimeId(value: string): string {
  return value.replace(/[^A-Za-z0-9_.-]/g, '_').slice(0, 120) || `worker-${process.pid}`
}

function runtimeIdentity() {
  const workerId = safeRuntimeId(
    process.env.TASK_DETAIL_UPLOAD_WORKER_ID
      || process.env.VIDEO_TRANSCODE_WORKER_ID
      || process.env.HOSTNAME
      || `worker-${process.pid}`,
  )
  const instanceId = safeRuntimeId(
    process.env.TASK_DETAIL_UPLOAD_INSTANCE_ID
      || process.env.VIDEO_TRANSCODE_INSTANCE_ID
      || `${workerId}-${process.pid}`,
  )
  const runtimeRoot = process.env.TASK_DETAIL_UPLOAD_RUNTIME_DIR
    || path.join(getUploadRootDir(), '.task-detail-upload-runtime')
  return { workerId, instanceId, runtimeRoot }
}

function capabilityPath() {
  const { workerId, runtimeRoot } = runtimeIdentity()
  return path.join(runtimeRoot, 'capabilities', `${workerId}.v1`)
}

function drainPath() {
  const { workerId, runtimeRoot } = runtimeIdentity()
  return path.join(runtimeRoot, 'drain', workerId)
}

function activePath() {
  const { workerId, runtimeRoot } = runtimeIdentity()
  return path.join(runtimeRoot, 'active', `${workerId}.json`)
}

function executionLockPath() {
  // 不跟随容器内的临时目录；UPLOAD_DIR 是蓝绿容器共享的 uploads 挂载。
  return path.join(getUploadRootDir(), '.locks', EXECUTION_LOCK_FILE_NAME)
}

async function atomicWriteFile(target: string, contents: string): Promise<void> {
  await mkdir(path.dirname(target), { recursive: true })
  const partialPath = `${target}.partial-${randomUUID()}`
  let handle: Awaited<ReturnType<typeof open>> | null = null
  try {
    handle = await open(partialPath, 'wx', 0o600)
    await handle.writeFile(contents, 'utf8')
    await handle.sync()
    await handle.close()
    handle = null
    if (process.platform === 'win32') {
      await rm(target, { force: true })
    }
    await rename(partialPath, target)
  } catch (error) {
    if (handle) {
      await handle.close().catch(() => undefined)
    }
    await rm(partialPath, { force: true }).catch(() => undefined)
    throw error
  }
}

export function getTaskDetailUploadWorkerIdentity() {
  const { workerId, instanceId } = runtimeIdentity()
  return { workerId, instanceId }
}

export async function ensureTaskDetailUploadDrainCapability(): Promise<void> {
  const { workerId, instanceId } = runtimeIdentity()
  await atomicWriteFile(
    capabilityPath(),
    `${TASK_DETAIL_UPLOAD_DRAIN_PROTOCOL}\nworkerId=${workerId}\ninstanceId=${instanceId}\n`,
  )
}

/**
 * 只有 ENOENT 才表示没有 drain。NAS、权限或 I/O 状态不明时必须保守暂停，
 * 否则部署期间可能在候选容器里悄悄领取长任务。
 */
export async function isTaskDetailUploadDraining(): Promise<boolean> {
  try {
    await access(drainPath())
    return true
  } catch (error) {
    if ((error as NodeJS.ErrnoException | undefined)?.code === 'ENOENT') {
      // Windows 对“父路径其实是文件”也可能返回 ENOENT。只有运行时
      // 路径结构本身正常时，才能把 ENOENT 解释为“没有 drain”。
      if (await hasSafeDrainPathStructure()) return false
    }
    console.error('[TaskDetailUploadRuntime] 无法确认 drain 标记，按暂停处理:', error)
    return true
  }
}

async function hasSafeDrainPathStructure(): Promise<boolean> {
  const { runtimeRoot } = runtimeIdentity()
  try {
    const rootStats = await lstat(runtimeRoot)
    if (!rootStats.isDirectory() || rootStats.isSymbolicLink()) return false
  } catch (error) {
    if ((error as NodeJS.ErrnoException | undefined)?.code === 'ENOENT') return true
    return false
  }

  try {
    const drainDirectoryStats = await lstat(path.dirname(drainPath()))
    return drainDirectoryStats.isDirectory() && !drainDirectoryStats.isSymbolicLink()
  } catch (error) {
    return (error as NodeJS.ErrnoException | undefined)?.code === 'ENOENT'
  }
}

export async function writeTaskDetailUploadActiveMarker(
  jobId: string,
  leaseToken: string,
): Promise<void> {
  const { workerId, instanceId } = runtimeIdentity()
  await atomicWriteFile(activePath(), JSON.stringify({
    protocol: TASK_DETAIL_UPLOAD_DRAIN_PROTOCOL,
    workerId,
    instanceId,
    jobId,
    leaseToken,
    pid: process.pid,
    heartbeatAt: new Date().toISOString(),
  }))
}

export async function removeTaskDetailUploadActiveMarker(): Promise<void> {
  await rm(activePath(), { force: true })
}

/**
 * 在真实的 Sharp/ZIP 生命周期外层持有独立 flock。数据库租约负责
 * “谁有权写结果”，该锁负责“全局同时只跑一个重处理”。Node 崩溃时
 * 持锁子进程的 stdin 会关闭，flock 随之自动释放。
 */
export async function acquireTaskDetailUploadExecutionLock(
  options: { timeoutMs?: number } = {},
): Promise<TaskDetailUploadExecutionLockRelease> {
  const timeoutMs = options.timeoutMs ?? DEFAULT_EXECUTION_LOCK_TIMEOUT_MS
  if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) {
    throw new TaskDetailUploadExecutionLockUnavailableError('任务详情执行锁超时参数无效')
  }

  if (process.platform !== 'linux' || process.env.NODE_ENV !== 'production') {
    return inProcessExecutionLock.acquire(timeoutMs)
  }

  const lockPath = executionLockPath()
  const lockDirectory = path.dirname(lockPath)
  await mkdir(lockDirectory, { recursive: true, mode: 0o750 })
  const directoryStats = await lstat(lockDirectory)
  if (!directoryStats.isDirectory() || directoryStats.isSymbolicLink()) {
    throw new TaskDetailUploadExecutionLockUnavailableError('任务详情执行锁目录异常')
  }

  try {
    const handle = await open(lockPath, 'wx', 0o600)
    await handle.close()
  } catch (error) {
    if ((error as NodeJS.ErrnoException | undefined)?.code !== 'EEXIST') throw error
  }
  const lockStats = await lstat(lockPath)
  if (!lockStats.isFile() || lockStats.isSymbolicLink()) {
    throw new TaskDetailUploadExecutionLockUnavailableError('任务详情执行锁文件异常')
  }

  const child = spawn(
    'flock',
    [
      '--exclusive',
      '--timeout',
      String(Math.max(1, Math.ceil(timeoutMs / 1_000))),
      '--conflict-exit-code',
      '75',
      '--no-fork',
      lockPath,
      'sh',
      '-c',
      `printf '${EXECUTION_LOCK_READY_MARKER}'; cat >/dev/null`,
    ],
    {
      shell: false,
      stdio: ['pipe', 'pipe', 'pipe'],
      windowsHide: true,
    },
  )
  return waitForExecutionFlock(child, timeoutMs)
}

function waitForExecutionFlock(
  child: ChildProcessWithoutNullStreams,
  timeoutMs: number,
): Promise<TaskDetailUploadExecutionLockRelease> {
  return new Promise((resolve, reject) => {
    let stdout = ''
    let stderr = ''
    let settled = false
    let released = false
    let lostError: Error | null = null
    const timer = setTimeout(() => {
      if (settled) return
      settled = true
      void terminateExecutionLockProcess(child).finally(() => {
        reject(new TaskDetailUploadExecutionLockUnavailableError())
      })
    }, timeoutMs + 1_000)
    timer.unref?.()

    const fail = (error: unknown) => {
      if (settled) return
      settled = true
      clearTimeout(timer)
      void terminateExecutionLockProcess(child).finally(() => {
        console.error('[TaskDetailUploadRuntime] execution flock 获取失败:', {
          error,
          stderr: stderr.slice(0, 1_000),
        })
        reject(new TaskDetailUploadExecutionLockUnavailableError())
      })
    }
    const recordLost = (source: string, error?: unknown) => {
      if (!settled || released || lostError) return
      lostError = new TaskDetailUploadExecutionLockLostError()
      console.error('[TaskDetailUploadRuntime] execution flock 意外丢失:', {
        source,
        error,
        stderr: stderr.slice(0, 1_000),
      })
    }

    child.stdin.on('error', (error) => recordLost('stdin', error))
    child.stdout.on('error', (error) => recordLost('stdout', error))
    child.stderr.on('error', (error) => recordLost('stderr', error))
    child.stderr.on('data', (chunk) => {
      stderr += String(chunk)
      if (stderr.length > 2_000) stderr = stderr.slice(-2_000)
    })
    child.on('error', (error) => settled ? recordLost('process', error) : fail(error))
    child.on('close', (code, signal) => {
      if (!settled) {
        fail(code === 75
          ? new TaskDetailUploadExecutionLockUnavailableError()
          : new Error(`flock 提前退出: code=${code}, signal=${signal}`))
      } else if (!released) {
        recordLost('close', new Error(`code=${code}, signal=${signal}`))
      }
    })
    child.stdout.on('data', (chunk) => {
      if (settled) return
      stdout += String(chunk)
      if (!stdout.includes(EXECUTION_LOCK_READY_MARKER)) {
        if (stdout.length > EXECUTION_LOCK_READY_MARKER.length * 4) {
          fail(new Error('flock 未返回预期就绪标记'))
        }
        return
      }

      settled = true
      clearTimeout(timer)
      const release = async () => {
        if (released) return
        released = true
        await terminateExecutionLockProcess(child)
      }
      release.assertHeld = () => {
        if (released || lostError || child.exitCode !== null || child.signalCode !== null) {
          throw lostError || new TaskDetailUploadExecutionLockLostError()
        }
      }
      resolve(release)
    })
  })
}

async function terminateExecutionLockProcess(child: ChildProcessWithoutNullStreams): Promise<void> {
  if (child.exitCode !== null || child.signalCode !== null) return
  const closed = new Promise<void>((resolve) => child.once('close', () => resolve()))
  child.stdin.end()
  if (await waitForExecutionLockPromise(closed, EXECUTION_LOCK_EXIT_TIMEOUT_MS)) return
  child.kill('SIGTERM')
  if (await waitForExecutionLockPromise(closed, EXECUTION_LOCK_EXIT_TIMEOUT_MS)) return
  child.kill('SIGKILL')
  await waitForExecutionLockPromise(closed, EXECUTION_LOCK_EXIT_TIMEOUT_MS)
}

async function waitForExecutionLockPromise(
  promise: Promise<void>,
  timeoutMs: number,
): Promise<boolean> {
  let timer: ReturnType<typeof setTimeout> | null = null
  const timeout = new Promise<false>((resolve) => {
    timer = setTimeout(() => resolve(false), timeoutMs)
    timer.unref?.()
  })
  const result = await Promise.race([promise.then(() => true), timeout])
  if (timer) clearTimeout(timer)
  return result
}

/** @internal 仅用于定向测试和部署协议断言。 */
export const __taskDetailUploadRuntimeTesting = {
  capabilityPath,
  drainPath,
  activePath,
  executionLockPath,
}