7326e273e1
Protect upload cleanup and atomic completion paths, add durable media worker leases and brand asset validation, and introduce verified migration/bootstrap/deployment gates.
579 lines
18 KiB
TypeScript
579 lines
18 KiB
TypeScript
import { spawn, type ChildProcessWithoutNullStreams } from 'child_process'
|
||
import { createHash } from 'crypto'
|
||
import { lstat, mkdir, open } from 'fs/promises'
|
||
import path from 'path'
|
||
|
||
import { MaterialUploadRequestError } from '@/lib/material-upload-request-guard'
|
||
import { getUploadRootDir } from '@/lib/upload-paths'
|
||
|
||
export type MaterialUploadStorageLockMode = 'shared' | 'exclusive'
|
||
export type MaterialUploadStorageLockRelease = (() => Promise<void>) & {
|
||
assertHeld: () => void
|
||
}
|
||
|
||
interface PendingLockRequest {
|
||
mode: MaterialUploadStorageLockMode
|
||
resolve: (release: MaterialUploadStorageLockRelease) => void
|
||
reject: (error: Error) => void
|
||
timer: ReturnType<typeof setTimeout>
|
||
}
|
||
|
||
const DEFAULT_LOCK_TIMEOUT_MS = 4 * 60 * 1000
|
||
const TASK_MATERIAL_LOCK_TIMEOUT_MS = 1_000
|
||
const TASK_MATERIAL_LOCK_HASH_LENGTH = 16
|
||
const FLOCK_EXIT_TIMEOUT_MS = 5_000
|
||
const LOCK_READY_MARKER = 'MATERIAL_UPLOAD_LOCK_READY\n'
|
||
|
||
/**
|
||
* Windows 本地开发和 Jest 使用的进程内读写锁。生产 Linux 必须使用下方的
|
||
* flock；生产环境绝不会静默降级到这个锁，否则蓝绿容器仍可同时写满上传盘。
|
||
*/
|
||
export class InProcessMaterialUploadReadWriteLock {
|
||
private activeReaders = 0
|
||
private writerActive = false
|
||
private readonly waiting: PendingLockRequest[] = []
|
||
|
||
acquire(
|
||
mode: MaterialUploadStorageLockMode,
|
||
timeoutMs = DEFAULT_LOCK_TIMEOUT_MS,
|
||
): Promise<MaterialUploadStorageLockRelease> {
|
||
if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) {
|
||
return Promise.reject(new Error('上传存储锁超时时间无效'))
|
||
}
|
||
|
||
if (this.canAcquireImmediately(mode)) {
|
||
return Promise.resolve(this.grant(mode))
|
||
}
|
||
|
||
return new Promise<MaterialUploadStorageLockRelease>((resolve, reject) => {
|
||
const request: PendingLockRequest = {
|
||
mode,
|
||
resolve,
|
||
reject,
|
||
timer: setTimeout(() => {
|
||
const index = this.waiting.indexOf(request)
|
||
if (index >= 0) {
|
||
this.waiting.splice(index, 1)
|
||
}
|
||
reject(createLockBusyError())
|
||
this.drainQueue()
|
||
}, timeoutMs),
|
||
}
|
||
request.timer.unref?.()
|
||
this.waiting.push(request)
|
||
})
|
||
}
|
||
|
||
snapshot(): { activeReaders: number; writerActive: boolean; waiting: number } {
|
||
return {
|
||
activeReaders: this.activeReaders,
|
||
writerActive: this.writerActive,
|
||
waiting: this.waiting.length,
|
||
}
|
||
}
|
||
|
||
private canAcquireImmediately(mode: MaterialUploadStorageLockMode): boolean {
|
||
if (mode === 'exclusive') {
|
||
return !this.writerActive && this.activeReaders === 0 && this.waiting.length === 0
|
||
}
|
||
|
||
// 已有写者排队后，新读者不能插队，否则持续分片流量会让合并永久饥饿。
|
||
return !this.writerActive && !this.waiting.some((request) => request.mode === 'exclusive')
|
||
}
|
||
|
||
private grant(mode: MaterialUploadStorageLockMode): MaterialUploadStorageLockRelease {
|
||
if (mode === 'exclusive') {
|
||
this.writerActive = true
|
||
} else {
|
||
this.activeReaders += 1
|
||
}
|
||
|
||
let released = false
|
||
const release = async () => {
|
||
if (released) {
|
||
return
|
||
}
|
||
released = true
|
||
if (mode === 'exclusive') {
|
||
this.writerActive = false
|
||
} else {
|
||
this.activeReaders = Math.max(0, this.activeReaders - 1)
|
||
}
|
||
this.drainQueue()
|
||
}
|
||
release.assertHeld = () => {
|
||
if (released) {
|
||
throw createLockLostError()
|
||
}
|
||
}
|
||
return release
|
||
}
|
||
|
||
private drainQueue(): void {
|
||
if (this.writerActive || this.waiting.length === 0) {
|
||
return
|
||
}
|
||
|
||
const first = this.waiting[0]
|
||
if (first.mode === 'exclusive') {
|
||
if (this.activeReaders !== 0) {
|
||
return
|
||
}
|
||
this.waiting.shift()
|
||
clearTimeout(first.timer)
|
||
first.resolve(this.grant('exclusive'))
|
||
return
|
||
}
|
||
|
||
while (this.waiting[0]?.mode === 'shared' && !this.writerActive) {
|
||
const reader = this.waiting.shift()!
|
||
clearTimeout(reader.timer)
|
||
reader.resolve(this.grant('shared'))
|
||
}
|
||
}
|
||
}
|
||
|
||
const inProcessLock = new InProcessMaterialUploadReadWriteLock()
|
||
const inProcessArchiveLock = new InProcessMaterialUploadReadWriteLock()
|
||
const inProcessTaskMaterialLocks = new Map<string, InProcessMaterialUploadReadWriteLock>()
|
||
const inProcessTaskUploadSessionLocks = new Map<string, InProcessMaterialUploadReadWriteLock>()
|
||
|
||
export async function acquireMaterialUploadStorageLock(
|
||
mode: MaterialUploadStorageLockMode,
|
||
options: { timeoutMs?: number } = {},
|
||
): Promise<MaterialUploadStorageLockRelease> {
|
||
const timeoutMs = options.timeoutMs ?? DEFAULT_LOCK_TIMEOUT_MS
|
||
if (process.platform !== 'linux' || process.env.NODE_ENV !== 'production') {
|
||
return inProcessLock.acquire(mode, timeoutMs)
|
||
}
|
||
|
||
return acquireLinuxFlock(mode, timeoutMs, 'material-upload-storage.lock')
|
||
}
|
||
|
||
/**
|
||
* ZIP 生成在蓝绿容器间仍必须串行，但不能用全局存储写锁堵住
|
||
* 其他用户已经写完的分片发布。独立 archive 锁保证同时只有一个
|
||
* 大 ZIP 写入；ZIP 另外持有全局 shared 锁，与分片共存，但会阻止
|
||
* complete/清理在 ZIP 还未结束时进入。
|
||
*/
|
||
export async function acquireMaterialUploadArchiveLock(
|
||
options: { timeoutMs?: number } = {},
|
||
): Promise<MaterialUploadStorageLockRelease> {
|
||
const timeoutMs = options.timeoutMs ?? DEFAULT_LOCK_TIMEOUT_MS
|
||
if (process.platform !== 'linux' || process.env.NODE_ENV !== 'production') {
|
||
return inProcessArchiveLock.acquire('exclusive', timeoutMs)
|
||
}
|
||
|
||
return acquireLinuxFlock('exclusive', timeoutMs, 'material-upload-archive.lock')
|
||
}
|
||
|
||
/**
|
||
* 同一条任务素材的完成动作必须跨蓝绿容器互斥，但不能占用全局上传存储锁：
|
||
* 完成流程内部还会获取全局锁来合并分片和生成 ZIP，复用同一把锁会自锁。
|
||
*/
|
||
export async function acquireTaskMaterialCompletionLock(
|
||
materialId: string,
|
||
options: { timeoutMs?: number } = {},
|
||
): Promise<MaterialUploadStorageLockRelease> {
|
||
const normalizedMaterialId = String(materialId || '').trim()
|
||
if (!/^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/.test(normalizedMaterialId)) {
|
||
throw new MaterialUploadRequestError(
|
||
'素材编号无效，无法开始完成处理',
|
||
400,
|
||
'TASK_MATERIAL_ID_INVALID',
|
||
)
|
||
}
|
||
|
||
const timeoutMs = options.timeoutMs ?? TASK_MATERIAL_LOCK_TIMEOUT_MS
|
||
if (process.platform !== 'linux' || process.env.NODE_ENV !== 'production') {
|
||
let lock = inProcessTaskMaterialLocks.get(normalizedMaterialId)
|
||
if (!lock) {
|
||
lock = new InProcessMaterialUploadReadWriteLock()
|
||
inProcessTaskMaterialLocks.set(normalizedMaterialId, lock)
|
||
}
|
||
try {
|
||
const innerRelease = await lock.acquire('exclusive', timeoutMs)
|
||
let released = false
|
||
const release = async () => {
|
||
if (released) {
|
||
return
|
||
}
|
||
released = true
|
||
await innerRelease()
|
||
const snapshot = lock!.snapshot()
|
||
if (!snapshot.writerActive && snapshot.activeReaders === 0 && snapshot.waiting === 0) {
|
||
inProcessTaskMaterialLocks.delete(normalizedMaterialId)
|
||
}
|
||
}
|
||
release.assertHeld = () => innerRelease.assertHeld()
|
||
return release
|
||
} catch (error) {
|
||
throw normalizeTaskMaterialLockError(error)
|
||
}
|
||
}
|
||
|
||
try {
|
||
return await acquireLinuxFlock(
|
||
'exclusive',
|
||
timeoutMs,
|
||
resolveScopedLockFileName('task-material', normalizedMaterialId),
|
||
)
|
||
} catch (error) {
|
||
throw normalizeTaskMaterialLockError(error)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 同一 uploadId 只能被一条任务素材完成流程消费。多个会话按固定顺序
|
||
* 取锁，防止不同 materialId 并发复用同一源文件时，一方清理掉另一方
|
||
* 正在读取的 assembled/temp 文件。
|
||
*/
|
||
export async function acquireTaskMaterialUploadSessionLocks(
|
||
uploadIds: string[],
|
||
options: { timeoutMs?: number } = {},
|
||
): Promise<MaterialUploadStorageLockRelease> {
|
||
const normalizedUploadIds = Array.from(new Set(
|
||
uploadIds.map((uploadId) => String(uploadId || '').trim()),
|
||
)).sort()
|
||
if (
|
||
normalizedUploadIds.length === 0
|
||
|| normalizedUploadIds.some((uploadId) => !/^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/.test(uploadId))
|
||
) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传会话编号无效，无法安全完成素材',
|
||
400,
|
||
'TASK_MATERIAL_UPLOAD_SESSION_INVALID',
|
||
)
|
||
}
|
||
|
||
const timeoutMs = options.timeoutMs ?? TASK_MATERIAL_LOCK_TIMEOUT_MS
|
||
const releases: MaterialUploadStorageLockRelease[] = []
|
||
try {
|
||
for (const uploadId of normalizedUploadIds) {
|
||
if (process.platform !== 'linux' || process.env.NODE_ENV !== 'production') {
|
||
let lock = inProcessTaskUploadSessionLocks.get(uploadId)
|
||
if (!lock) {
|
||
lock = new InProcessMaterialUploadReadWriteLock()
|
||
inProcessTaskUploadSessionLocks.set(uploadId, lock)
|
||
}
|
||
const innerRelease = await lock.acquire('exclusive', timeoutMs)
|
||
const release = Object.assign(async () => {
|
||
await innerRelease()
|
||
const snapshot = lock!.snapshot()
|
||
if (!snapshot.writerActive && snapshot.activeReaders === 0 && snapshot.waiting === 0) {
|
||
inProcessTaskUploadSessionLocks.delete(uploadId)
|
||
}
|
||
}, {
|
||
assertHeld: () => innerRelease.assertHeld(),
|
||
})
|
||
releases.push(release)
|
||
} else {
|
||
releases.push(await acquireLinuxFlock(
|
||
'exclusive',
|
||
timeoutMs,
|
||
resolveScopedLockFileName('task-upload', uploadId),
|
||
))
|
||
}
|
||
}
|
||
} catch (error) {
|
||
for (const release of releases.reverse()) {
|
||
await release().catch(() => undefined)
|
||
}
|
||
throw normalizeTaskUploadSessionLockError(error)
|
||
}
|
||
|
||
let released = false
|
||
const releaseAll = async () => {
|
||
if (released) {
|
||
return
|
||
}
|
||
released = true
|
||
for (const release of releases.reverse()) {
|
||
await release()
|
||
}
|
||
}
|
||
releaseAll.assertHeld = () => {
|
||
if (released) {
|
||
throw createLockLostError()
|
||
}
|
||
releases.forEach((release) => release.assertHeld())
|
||
}
|
||
return releaseAll
|
||
}
|
||
|
||
export function resolveScopedLockFileName(scope: string, key: string): string {
|
||
const normalizedScope = String(scope || '').trim()
|
||
if (!/^[A-Za-z0-9][A-Za-z0-9_-]{0,31}$/.test(normalizedScope)) {
|
||
throw new Error('上传锁作用域无效')
|
||
}
|
||
const digest = createHash('sha256')
|
||
.update(String(key || '').trim())
|
||
.digest('hex')
|
||
.slice(0, TASK_MATERIAL_LOCK_HASH_LENGTH)
|
||
return `${normalizedScope}-${digest}.lock`
|
||
}
|
||
|
||
async function acquireLinuxFlock(
|
||
mode: MaterialUploadStorageLockMode,
|
||
timeoutMs: number,
|
||
lockFileName: string,
|
||
): Promise<MaterialUploadStorageLockRelease> {
|
||
const lockPath = await prepareLockFile(lockFileName)
|
||
const timeoutSeconds = Math.max(1, Math.ceil(timeoutMs / 1000))
|
||
const child = spawn(
|
||
'flock',
|
||
[
|
||
mode === 'shared' ? '--shared' : '--exclusive',
|
||
'--timeout',
|
||
String(timeoutSeconds),
|
||
'--conflict-exit-code',
|
||
'75',
|
||
'--no-fork',
|
||
lockPath,
|
||
'sh',
|
||
'-c',
|
||
`printf '${LOCK_READY_MARKER}'; cat >/dev/null`,
|
||
],
|
||
{
|
||
shell: false,
|
||
stdio: ['pipe', 'pipe', 'pipe'],
|
||
windowsHide: true,
|
||
},
|
||
)
|
||
|
||
return waitForFlockReady(child, timeoutMs)
|
||
}
|
||
|
||
async function prepareLockFile(lockFileName: string): Promise<string> {
|
||
if (!/^[A-Za-z0-9][A-Za-z0-9_.-]{0,191}$/.test(lockFileName)) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传存储锁名称无效',
|
||
503,
|
||
'UPLOAD_STORAGE_LOCK_UNAVAILABLE',
|
||
3,
|
||
)
|
||
}
|
||
const uploadRoot = getUploadRootDir()
|
||
const lockDirectory = path.join(uploadRoot, '.locks')
|
||
const lockPath = path.join(lockDirectory, lockFileName)
|
||
await mkdir(lockDirectory, { recursive: true, mode: 0o750 })
|
||
|
||
const directoryStats = await lstat(lockDirectory)
|
||
if (!directoryStats.isDirectory() || directoryStats.isSymbolicLink()) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传存储锁目录异常，请联系管理员',
|
||
503,
|
||
'UPLOAD_STORAGE_LOCK_UNAVAILABLE',
|
||
3,
|
||
)
|
||
}
|
||
|
||
try {
|
||
const handle = await open(lockPath, 'wx', 0o600)
|
||
await handle.close()
|
||
} catch (error: any) {
|
||
if (error?.code !== 'EEXIST') {
|
||
throw error
|
||
}
|
||
}
|
||
|
||
const lockStats = await lstat(lockPath)
|
||
if (!lockStats.isFile() || lockStats.isSymbolicLink()) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传存储锁文件异常，请联系管理员',
|
||
503,
|
||
'UPLOAD_STORAGE_LOCK_UNAVAILABLE',
|
||
3,
|
||
)
|
||
}
|
||
return lockPath
|
||
}
|
||
|
||
function waitForFlockReady(
|
||
child: ChildProcessWithoutNullStreams,
|
||
timeoutMs: number,
|
||
): Promise<MaterialUploadStorageLockRelease> {
|
||
return new Promise((resolve, reject) => {
|
||
let stdout = ''
|
||
let stderr = ''
|
||
let settled = false
|
||
let released = false
|
||
let lostError: Error | null = null
|
||
const timeout = setTimeout(() => {
|
||
if (settled) {
|
||
return
|
||
}
|
||
settled = true
|
||
terminateLockProcess(child).finally(() => reject(createLockBusyError()))
|
||
}, timeoutMs + 1_000)
|
||
timeout.unref?.()
|
||
|
||
const fail = (error: unknown) => {
|
||
if (settled) {
|
||
return
|
||
}
|
||
settled = true
|
||
clearTimeout(timeout)
|
||
terminateLockProcess(child).finally(() => {
|
||
console.error('[MaterialUploadLock] flock 获取失败:', {
|
||
error,
|
||
stderr: stderr.slice(0, 1_000),
|
||
})
|
||
reject(
|
||
error instanceof MaterialUploadRequestError
|
||
? error
|
||
: new MaterialUploadRequestError(
|
||
'上传存储正忙或锁服务不可用，请稍后重试',
|
||
503,
|
||
'UPLOAD_STORAGE_LOCK_UNAVAILABLE',
|
||
3,
|
||
),
|
||
)
|
||
})
|
||
}
|
||
|
||
const recordLostLock = (source: string, error?: unknown) => {
|
||
if (!settled || released || lostError) {
|
||
return
|
||
}
|
||
lostError = createLockLostError()
|
||
console.error('[MaterialUploadLock] 已获取的 flock 意外丢失:', {
|
||
source,
|
||
error,
|
||
stderr: stderr.slice(0, 1_000),
|
||
})
|
||
}
|
||
|
||
child.stdin.on('error', (error) => recordLostLock('stdin', error))
|
||
child.stdout.on('error', (error) => recordLostLock('stdout', error))
|
||
child.stderr.on('error', (error) => recordLostLock('stderr', error))
|
||
child.stderr.on('data', (chunk) => {
|
||
stderr += String(chunk)
|
||
if (stderr.length > 2_000) {
|
||
stderr = stderr.slice(-2_000)
|
||
}
|
||
})
|
||
child.on('error', (error) => {
|
||
if (settled) {
|
||
recordLostLock('process', error)
|
||
} else {
|
||
fail(error)
|
||
}
|
||
})
|
||
child.on('close', (code, signal) => {
|
||
if (!settled) {
|
||
fail(
|
||
code === 75
|
||
? createLockBusyError()
|
||
: new Error(`flock 提前退出: code=${code}, signal=${signal}`),
|
||
)
|
||
} else if (!released) {
|
||
recordLostLock('close', new Error(`code=${code}, signal=${signal}`))
|
||
}
|
||
})
|
||
child.stdout.on('data', (chunk) => {
|
||
if (settled) {
|
||
return
|
||
}
|
||
stdout += String(chunk)
|
||
if (!stdout.includes(LOCK_READY_MARKER)) {
|
||
if (stdout.length > LOCK_READY_MARKER.length * 4) {
|
||
fail(new Error('flock 未返回预期就绪标记'))
|
||
}
|
||
return
|
||
}
|
||
|
||
settled = true
|
||
clearTimeout(timeout)
|
||
const release = async () => {
|
||
if (released) {
|
||
return
|
||
}
|
||
released = true
|
||
await terminateLockProcess(child)
|
||
}
|
||
release.assertHeld = () => {
|
||
if (released || lostError || child.exitCode !== null || child.signalCode !== null) {
|
||
throw lostError || createLockLostError()
|
||
}
|
||
}
|
||
resolve(release)
|
||
})
|
||
})
|
||
}
|
||
|
||
async function terminateLockProcess(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 waitForPromise(closed, FLOCK_EXIT_TIMEOUT_MS)) {
|
||
return
|
||
}
|
||
|
||
child.kill('SIGTERM')
|
||
if (await waitForPromise(closed, FLOCK_EXIT_TIMEOUT_MS)) {
|
||
return
|
||
}
|
||
child.kill('SIGKILL')
|
||
await waitForPromise(closed, FLOCK_EXIT_TIMEOUT_MS)
|
||
}
|
||
|
||
async function waitForPromise(promise: Promise<void>, timeoutMs: number): Promise<boolean> {
|
||
let timer: ReturnType<typeof setTimeout> | undefined
|
||
const timedOut = new Promise<false>((resolve) => {
|
||
timer = setTimeout(() => resolve(false), timeoutMs)
|
||
timer.unref?.()
|
||
})
|
||
const completed = promise.then(() => true)
|
||
const result = await Promise.race([completed, timedOut])
|
||
if (timer) {
|
||
clearTimeout(timer)
|
||
}
|
||
return result
|
||
}
|
||
|
||
function createLockBusyError(): MaterialUploadRequestError {
|
||
return new MaterialUploadRequestError(
|
||
'上传存储正在处理其它大文件，请稍后重试',
|
||
503,
|
||
'UPLOAD_STORAGE_BUSY',
|
||
3,
|
||
)
|
||
}
|
||
|
||
function createLockLostError(): MaterialUploadRequestError {
|
||
return new MaterialUploadRequestError(
|
||
'上传存储锁意外中断，本次写入已安全取消，请重试',
|
||
503,
|
||
'UPLOAD_STORAGE_LOCK_LOST',
|
||
3,
|
||
)
|
||
}
|
||
|
||
function normalizeTaskMaterialLockError(error: unknown): Error {
|
||
if (error instanceof MaterialUploadRequestError && error.code === 'UPLOAD_STORAGE_BUSY') {
|
||
return new MaterialUploadRequestError(
|
||
'该素材正在处理，请勿重复提交',
|
||
409,
|
||
'TASK_MATERIAL_COMPLETION_BUSY',
|
||
3,
|
||
)
|
||
}
|
||
return error instanceof Error ? error : new Error(String(error))
|
||
}
|
||
|
||
function normalizeTaskUploadSessionLockError(error: unknown): Error {
|
||
if (error instanceof MaterialUploadRequestError && error.code === 'UPLOAD_STORAGE_BUSY') {
|
||
return new MaterialUploadRequestError(
|
||
'上传文件正被另一条素材完成处理，请稍后重试',
|
||
409,
|
||
'TASK_MATERIAL_UPLOAD_SESSION_BUSY',
|
||
1,
|
||
)
|
||
}
|
||
return error instanceof Error ? error : new Error(String(error))
|
||
}
|