7326e273e1
Protect upload cleanup and atomic completion paths, add durable media worker leases and brand asset validation, and introduce verified migration/bootstrap/deployment gates.
277 lines
9.1 KiB
TypeScript
277 lines
9.1 KiB
TypeScript
import { spawn, type ChildProcessWithoutNullStreams } from 'child_process'
|
||
import { lstat, mkdir, open } from 'fs/promises'
|
||
import path from 'path'
|
||
import { AsyncLocalStorage } from 'async_hooks'
|
||
|
||
import { getUploadRootDir } from '@/lib/upload-paths'
|
||
|
||
const DEFAULT_MAX_CONCURRENT_IMAGE_JOBS = 1
|
||
const MAX_ALLOWED_IMAGE_JOBS = 2
|
||
const DEFAULT_IMAGE_LOCK_TIMEOUT_MS = 60_000
|
||
const IMAGE_LOCK_EXIT_TIMEOUT_MS = 5_000
|
||
const IMAGE_LOCK_READY_MARKER = 'IMAGE_PROCESSING_LOCK_READY\n'
|
||
const IMAGE_LOCK_FILE_NAME = 'image-processing-global.lock'
|
||
|
||
export class ImageProcessingLockUnavailableError extends Error {
|
||
constructor(message = '图片处理服务正忙，请稍后重试') {
|
||
super(message)
|
||
this.name = 'ImageProcessingLockUnavailableError'
|
||
}
|
||
}
|
||
|
||
export class ImageProcessingLockLostError extends Error {
|
||
constructor() {
|
||
super('图片处理锁意外中断，本次结果已安全拒绝')
|
||
this.name = 'ImageProcessingLockLostError'
|
||
}
|
||
}
|
||
|
||
type ImageProcessingLockRelease = (() => Promise<void>) & {
|
||
assertHeld: () => void
|
||
}
|
||
|
||
const imageProcessingLockContext = new AsyncLocalStorage<{
|
||
assertHeld: () => void
|
||
}>()
|
||
|
||
function readMaxConcurrentImageJobs(): number {
|
||
const configured = Number.parseInt(process.env.IMAGE_PROCESSING_MAX_CONCURRENT || '', 10)
|
||
if (!Number.isFinite(configured)) {
|
||
return DEFAULT_MAX_CONCURRENT_IMAGE_JOBS
|
||
}
|
||
|
||
return Math.max(1, Math.min(MAX_ALLOWED_IMAGE_JOBS, configured))
|
||
}
|
||
|
||
const maxConcurrentImageJobs = readMaxConcurrentImageJobs()
|
||
let activeImageJobs = 0
|
||
const waitingImageJobs: Array<() => void> = []
|
||
|
||
async function acquireInProcessImageProcessingSlot(): Promise<ImageProcessingLockRelease> {
|
||
if (activeImageJobs >= maxConcurrentImageJobs) {
|
||
await new Promise<void>((resolve) => waitingImageJobs.push(resolve))
|
||
} else {
|
||
activeImageJobs += 1
|
||
}
|
||
|
||
let released = false
|
||
const release = async () => {
|
||
if (released) return
|
||
released = true
|
||
const next = waitingImageJobs.shift()
|
||
if (next) {
|
||
// 当前名额直接交给队首等待者，避免释放与重新获取之间被后来请求插队。
|
||
next()
|
||
return
|
||
}
|
||
activeImageJobs = Math.max(0, activeImageJobs - 1)
|
||
}
|
||
release.assertHeld = () => {
|
||
if (released) throw new ImageProcessingLockLostError()
|
||
}
|
||
return release
|
||
}
|
||
|
||
function imageProcessingLockPath(): string {
|
||
// UPLOAD_DIR 是蓝绿容器共同挂载的目录，不能使用各容器自己的 /tmp。
|
||
return path.join(getUploadRootDir(), '.locks', IMAGE_LOCK_FILE_NAME)
|
||
}
|
||
|
||
async function prepareImageProcessingLockFile(): Promise<string> {
|
||
const lockPath = imageProcessingLockPath()
|
||
const lockDirectory = path.dirname(lockPath)
|
||
await mkdir(lockDirectory, { recursive: true, mode: 0o750 })
|
||
|
||
const directoryStats = await lstat(lockDirectory)
|
||
if (!directoryStats.isDirectory() || directoryStats.isSymbolicLink()) {
|
||
throw new ImageProcessingLockUnavailableError('图片处理锁目录异常，请联系管理员')
|
||
}
|
||
|
||
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 ImageProcessingLockUnavailableError('图片处理锁文件异常，请联系管理员')
|
||
}
|
||
return lockPath
|
||
}
|
||
|
||
async function acquireLinuxImageProcessingLock(): Promise<ImageProcessingLockRelease> {
|
||
const lockPath = await prepareImageProcessingLockFile()
|
||
const child = spawn(
|
||
'flock',
|
||
[
|
||
'--exclusive',
|
||
'--timeout',
|
||
String(Math.ceil(DEFAULT_IMAGE_LOCK_TIMEOUT_MS / 1_000)),
|
||
'--conflict-exit-code',
|
||
'75',
|
||
'--no-fork',
|
||
lockPath,
|
||
'sh',
|
||
'-c',
|
||
`printf '${IMAGE_LOCK_READY_MARKER}'; cat >/dev/null`,
|
||
],
|
||
{
|
||
shell: false,
|
||
stdio: ['pipe', 'pipe', 'pipe'],
|
||
windowsHide: true,
|
||
},
|
||
)
|
||
return waitForImageProcessingFlock(child)
|
||
}
|
||
|
||
function waitForImageProcessingFlock(
|
||
child: ChildProcessWithoutNullStreams,
|
||
): Promise<ImageProcessingLockRelease> {
|
||
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 terminateImageLockProcess(child).finally(() => {
|
||
reject(new ImageProcessingLockUnavailableError())
|
||
})
|
||
}, DEFAULT_IMAGE_LOCK_TIMEOUT_MS + 1_000)
|
||
timer.unref?.()
|
||
|
||
const fail = (error: unknown) => {
|
||
if (settled) return
|
||
settled = true
|
||
clearTimeout(timer)
|
||
void terminateImageLockProcess(child).finally(() => {
|
||
console.error('[ImageProcessingLimiter] flock 获取失败:', {
|
||
error,
|
||
stderr: stderr.slice(0, 1_000),
|
||
})
|
||
reject(new ImageProcessingLockUnavailableError())
|
||
})
|
||
}
|
||
const recordLost = (source: string, error?: unknown) => {
|
||
if (!settled || released || lostError) return
|
||
lostError = new ImageProcessingLockLostError()
|
||
console.error('[ImageProcessingLimiter] 已获取的 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 ImageProcessingLockUnavailableError()
|
||
: 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(IMAGE_LOCK_READY_MARKER)) {
|
||
if (stdout.length > IMAGE_LOCK_READY_MARKER.length * 4) {
|
||
fail(new Error('flock 未返回预期就绪标记'))
|
||
}
|
||
return
|
||
}
|
||
|
||
settled = true
|
||
clearTimeout(timer)
|
||
const release = async () => {
|
||
if (released) return
|
||
released = true
|
||
await terminateImageLockProcess(child)
|
||
}
|
||
release.assertHeld = () => {
|
||
if (released || lostError || child.exitCode !== null || child.signalCode !== null) {
|
||
throw lostError || new ImageProcessingLockLostError()
|
||
}
|
||
}
|
||
resolve(release)
|
||
})
|
||
})
|
||
}
|
||
|
||
async function terminateImageLockProcess(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 waitForImageLockPromise(closed, IMAGE_LOCK_EXIT_TIMEOUT_MS)) return
|
||
child.kill('SIGTERM')
|
||
if (await waitForImageLockPromise(closed, IMAGE_LOCK_EXIT_TIMEOUT_MS)) return
|
||
child.kill('SIGKILL')
|
||
await waitForImageLockPromise(closed, IMAGE_LOCK_EXIT_TIMEOUT_MS)
|
||
}
|
||
|
||
async function waitForImageLockPromise(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
|
||
}
|
||
|
||
async function acquireImageProcessingSlot(): Promise<ImageProcessingLockRelease> {
|
||
if (process.platform === 'linux' && process.env.NODE_ENV === 'production') {
|
||
// 生产环境固定整机并发 1；不得用环境变量把蓝绿总并发放大。
|
||
return acquireLinuxImageProcessingLock()
|
||
}
|
||
return acquireInProcessImageProcessingSlot()
|
||
}
|
||
|
||
/**
|
||
* 所有 Sharp 重处理共用的门闩。生产 Linux 会在共享 uploads 上持有独立 OS
|
||
* flock；进程崩溃后内核自动释放。任务完成前后都确认锁仍在，锁状态不明时
|
||
* 失败关闭，调用方不得提交本次派生结果。
|
||
*/
|
||
export async function withImageProcessingSlot<T>(task: () => Promise<T>): Promise<T> {
|
||
const inheritedLock = imageProcessingLockContext.getStore()
|
||
if (inheritedLock) {
|
||
// image-processor 的高层函数可能在已持锁的安全流程内调用另一个受限函数。
|
||
// 同一异步调用链复用锁，既避免自锁，也不允许脱离父任务后继续使用旧锁。
|
||
inheritedLock.assertHeld()
|
||
const result = await task()
|
||
inheritedLock.assertHeld()
|
||
return result
|
||
}
|
||
|
||
const release = await acquireImageProcessingSlot()
|
||
try {
|
||
release.assertHeld()
|
||
const result = await imageProcessingLockContext.run(
|
||
{ assertHeld: release.assertHeld },
|
||
task,
|
||
)
|
||
release.assertHeld()
|
||
return result
|
||
} finally {
|
||
await release()
|
||
}
|
||
}
|
||
|
||
/** @internal 仅供定向测试确认共享锁位置。 */
|
||
export const __imageProcessingLimiterTesting = {
|
||
imageProcessingLockPath,
|
||
}
|