7326e273e1
Protect upload cleanup and atomic completion paths, add durable media worker leases and brand asset validation, and introduce verified migration/bootstrap/deployment gates.
391 lines
11 KiB
TypeScript
391 lines
11 KiB
TypeScript
import { randomUUID } from 'crypto'
|
||
import { lstat, open, readFile, readdir, rename, rm } from 'fs/promises'
|
||
import path from 'path'
|
||
|
||
import { MaterialUploadRequestError } from '@/lib/material-upload-request-guard'
|
||
import { getUploadRootDir } from '@/lib/upload-paths'
|
||
import {
|
||
ensureSafeDirectoryWithinRoot,
|
||
resolveStrictChildPath,
|
||
} from '@/lib/path-security'
|
||
|
||
const SAFE_UPLOAD_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/
|
||
const COMPLETED_SEAL_RETENTION_MS = 24 * 60 * 60 * 1000
|
||
|
||
interface StoredUploadCompletionSeal {
|
||
version: 1
|
||
state: 'sealing' | 'completed' | 'cancelled'
|
||
token: string
|
||
createdAt: number
|
||
completedAt?: number
|
||
cancelledAt?: number
|
||
}
|
||
|
||
export type UploadSessionSealState = StoredUploadCompletionSeal['state']
|
||
|
||
export interface UploadSessionSealSnapshot {
|
||
state: UploadSessionSealState
|
||
createdAt: number
|
||
completedAt?: number
|
||
cancelledAt?: number
|
||
}
|
||
|
||
export interface UploadCompletionSealHandle {
|
||
uploadId: string
|
||
token: string
|
||
}
|
||
|
||
export type BeginUploadCompletionSealResult =
|
||
| { state: 'completed' }
|
||
| { state: 'sealing'; handle: UploadCompletionSealHandle }
|
||
|
||
/**
|
||
* 必须在 material upload 的跨容器独占锁内调用。
|
||
*
|
||
* sealing 标记会先于最终文件发布而写入共享 uploads。若进程在原子 rename 后、
|
||
* 写 completed 前崩溃，迟到分片仍会被挡住；下一次 complete 取得独占锁后可
|
||
* 根据最终文件继续收口，不会把旧会话复活。
|
||
*/
|
||
export async function beginUploadCompletionSeal(
|
||
uploadId: string,
|
||
): Promise<BeginUploadCompletionSealResult> {
|
||
const normalizedUploadId = normalizeUploadId(uploadId)
|
||
const existing = await readStoredSeal(normalizedUploadId)
|
||
if (existing?.state === 'cancelled') {
|
||
throw createUploadSessionCancelledError()
|
||
}
|
||
if (existing?.state === 'completed') {
|
||
return { state: 'completed' }
|
||
}
|
||
|
||
const token = randomUUID()
|
||
await writeStoredSeal(normalizedUploadId, {
|
||
version: 1,
|
||
state: 'sealing',
|
||
token,
|
||
createdAt: Date.now(),
|
||
})
|
||
return {
|
||
state: 'sealing',
|
||
handle: { uploadId: normalizedUploadId, token },
|
||
}
|
||
}
|
||
|
||
export async function completeUploadCompletionSeal(
|
||
handle: UploadCompletionSealHandle,
|
||
): Promise<void> {
|
||
const uploadId = normalizeUploadId(handle.uploadId)
|
||
const current = await readStoredSeal(uploadId)
|
||
if (current?.state === 'cancelled') {
|
||
throw createUploadSessionCancelledError()
|
||
}
|
||
if (!current || current.state !== 'sealing' || current.token !== handle.token) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传完成标记已变化，已停止发布结果',
|
||
409,
|
||
'UPLOAD_SESSION_SEAL_CHANGED',
|
||
)
|
||
}
|
||
|
||
await writeStoredSeal(uploadId, {
|
||
...current,
|
||
state: 'completed',
|
||
completedAt: Date.now(),
|
||
})
|
||
}
|
||
|
||
export async function abortUploadCompletionSeal(
|
||
handle: UploadCompletionSealHandle,
|
||
): Promise<void> {
|
||
const uploadId = normalizeUploadId(handle.uploadId)
|
||
const current = await readStoredSeal(uploadId).catch(() => null)
|
||
if (!current || current.state !== 'sealing' || current.token !== handle.token) {
|
||
return
|
||
}
|
||
|
||
const sealPath = await resolveSealPath(uploadId, false)
|
||
if (sealPath) {
|
||
await rm(sealPath, { force: true })
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 分片在短共享锁内、发布 chunk-N 之前调用。任何 sealing/completed 标记都
|
||
* 代表完成流程已经接管该 uploadId；此时必须删除本请求 partial 并拒绝更新
|
||
* 内存/Redis 会话。
|
||
*/
|
||
export async function assertUploadSessionAcceptingChunks(uploadId: string): Promise<void> {
|
||
const normalizedUploadId = normalizeUploadId(uploadId)
|
||
const seal = await readStoredSeal(normalizedUploadId)
|
||
if (seal?.state === 'cancelled') {
|
||
throw createUploadSessionCancelledError()
|
||
}
|
||
if (seal) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传会话已经进入完成阶段，迟到分片已拒绝',
|
||
409,
|
||
'UPLOAD_SESSION_SEALED',
|
||
)
|
||
}
|
||
}
|
||
|
||
/**
|
||
* cleanup 在 material upload 独占锁内调用。cancelled 是持久终态：一旦写入，
|
||
* complete 和迟到 chunk 都只能结构化失败，不能把已经取消的 uploadId 复活。
|
||
*/
|
||
export async function cancelUploadSessionSeal(
|
||
uploadId: string,
|
||
): Promise<{ state: 'cancelled' } | { state: 'completed' }> {
|
||
const normalizedUploadId = normalizeUploadId(uploadId)
|
||
const existing = await readStoredSeal(normalizedUploadId)
|
||
if (existing?.state === 'completed') {
|
||
return { state: 'completed' }
|
||
}
|
||
if (existing?.state === 'cancelled') {
|
||
return { state: 'cancelled' }
|
||
}
|
||
|
||
const now = Date.now()
|
||
await writeStoredSeal(normalizedUploadId, {
|
||
version: 1,
|
||
state: 'cancelled',
|
||
token: randomUUID(),
|
||
createdAt: existing?.createdAt ?? now,
|
||
cancelledAt: now,
|
||
})
|
||
return { state: 'cancelled' }
|
||
}
|
||
|
||
export async function readUploadSessionSeal(
|
||
uploadId: string,
|
||
): Promise<UploadSessionSealSnapshot | null> {
|
||
const seal = await readStoredSeal(normalizeUploadId(uploadId))
|
||
return seal
|
||
? {
|
||
state: seal.state,
|
||
createdAt: seal.createdAt,
|
||
completedAt: seal.completedAt,
|
||
cancelledAt: seal.cancelledAt,
|
||
}
|
||
: null
|
||
}
|
||
|
||
export async function cleanupExpiredUploadCompletionSeals(
|
||
now = Date.now(),
|
||
): Promise<number> {
|
||
const sealDir = await resolveSealDirectory(false)
|
||
if (!sealDir) {
|
||
return 0
|
||
}
|
||
|
||
let names: string[]
|
||
try {
|
||
names = await readdir(sealDir)
|
||
} catch (error: any) {
|
||
if (error?.code === 'ENOENT') {
|
||
return 0
|
||
}
|
||
throw error
|
||
}
|
||
|
||
let removed = 0
|
||
for (const name of names) {
|
||
if (!name.endsWith('.json')) {
|
||
continue
|
||
}
|
||
const uploadId = name.slice(0, -'.json'.length)
|
||
if (!SAFE_UPLOAD_ID_PATTERN.test(uploadId)) {
|
||
continue
|
||
}
|
||
const sealPath = resolveStrictChildPath(sealDir, name)
|
||
if (!sealPath) {
|
||
continue
|
||
}
|
||
try {
|
||
const stats = await lstat(sealPath)
|
||
if (
|
||
stats.isSymbolicLink()
|
||
|| !stats.isFile()
|
||
|| now - stats.mtimeMs < COMPLETED_SEAL_RETENTION_MS
|
||
) {
|
||
continue
|
||
}
|
||
const seal = await readStoredSeal(uploadId)
|
||
if (!seal || seal.state === 'sealing') {
|
||
continue
|
||
}
|
||
// 终态标记只有在对应临时目录已经不存在后才可过期。否则先删 seal 会让
|
||
// 下一轮自动清理把仍待业务消费的 completed merged 文件误判为孤儿。
|
||
const uploadTempPath = resolveStrictChildPath(
|
||
getUploadRootDir(),
|
||
'temp',
|
||
uploadId,
|
||
)
|
||
if (uploadTempPath) {
|
||
try {
|
||
await lstat(uploadTempPath)
|
||
continue
|
||
} catch (error: any) {
|
||
if (error?.code !== 'ENOENT') {
|
||
throw error
|
||
}
|
||
}
|
||
}
|
||
await rm(sealPath, { force: true })
|
||
removed += 1
|
||
} catch (error: any) {
|
||
if (error?.code !== 'ENOENT') {
|
||
console.warn('[UploadSessionSeal] 清理过期完成标记失败:', { sealPath, error })
|
||
}
|
||
}
|
||
}
|
||
return removed
|
||
}
|
||
|
||
function normalizeUploadId(uploadId: string): string {
|
||
const normalized = String(uploadId || '').trim()
|
||
if (!SAFE_UPLOAD_ID_PATTERN.test(normalized)) {
|
||
throw new MaterialUploadRequestError('uploadId 无效', 400, 'UPLOAD_ID_INVALID')
|
||
}
|
||
return normalized
|
||
}
|
||
|
||
async function resolveSealDirectory(create: boolean): Promise<string | null> {
|
||
const uploadRoot = getUploadRootDir()
|
||
if (create) {
|
||
return ensureSafeDirectoryWithinRoot(
|
||
uploadRoot,
|
||
'.locks',
|
||
'completed-upload-sessions',
|
||
)
|
||
}
|
||
|
||
const sealDir = resolveStrictChildPath(uploadRoot, '.locks', 'completed-upload-sessions')
|
||
if (!sealDir) {
|
||
return null
|
||
}
|
||
try {
|
||
const stats = await lstat(sealDir)
|
||
if (stats.isSymbolicLink() || !stats.isDirectory()) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传完成标记目录异常，已停止写入',
|
||
503,
|
||
'UPLOAD_SESSION_SEAL_INVALID',
|
||
)
|
||
}
|
||
return sealDir
|
||
} catch (error: any) {
|
||
if (error?.code === 'ENOENT') {
|
||
return null
|
||
}
|
||
throw error
|
||
}
|
||
}
|
||
|
||
async function resolveSealPath(uploadId: string, createDirectory: boolean): Promise<string | null> {
|
||
const sealDir = await resolveSealDirectory(createDirectory)
|
||
return sealDir ? resolveStrictChildPath(sealDir, `${uploadId}.json`) : null
|
||
}
|
||
|
||
async function readStoredSeal(uploadId: string): Promise<StoredUploadCompletionSeal | null> {
|
||
const sealPath = await resolveSealPath(uploadId, false)
|
||
if (!sealPath) {
|
||
return null
|
||
}
|
||
try {
|
||
const stats = await lstat(sealPath)
|
||
if (stats.isSymbolicLink() || !stats.isFile() || stats.size > 4096) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传完成标记异常，已停止写入',
|
||
503,
|
||
'UPLOAD_SESSION_SEAL_INVALID',
|
||
)
|
||
}
|
||
const parsed = JSON.parse(await readFile(sealPath, 'utf8')) as Partial<StoredUploadCompletionSeal>
|
||
if (
|
||
parsed.version !== 1
|
||
|| (
|
||
parsed.state !== 'sealing'
|
||
&& parsed.state !== 'completed'
|
||
&& parsed.state !== 'cancelled'
|
||
)
|
||
|| typeof parsed.token !== 'string'
|
||
|| !parsed.token
|
||
|| !Number.isFinite(parsed.createdAt)
|
||
|| (parsed.state === 'cancelled' && !Number.isFinite(parsed.cancelledAt))
|
||
) {
|
||
throw new MaterialUploadRequestError(
|
||
'上传完成标记异常，已停止写入',
|
||
503,
|
||
'UPLOAD_SESSION_SEAL_INVALID',
|
||
)
|
||
}
|
||
return parsed as StoredUploadCompletionSeal
|
||
} catch (error: any) {
|
||
if (error?.code === 'ENOENT') {
|
||
return null
|
||
}
|
||
if (error instanceof MaterialUploadRequestError) {
|
||
throw error
|
||
}
|
||
throw new MaterialUploadRequestError(
|
||
'无法确认上传会话完成状态，已停止写入',
|
||
503,
|
||
'UPLOAD_SESSION_SEAL_UNAVAILABLE',
|
||
)
|
||
}
|
||
}
|
||
|
||
function createUploadSessionCancelledError(): MaterialUploadRequestError {
|
||
return new MaterialUploadRequestError(
|
||
'上传会话已取消，不能继续上传或完成',
|
||
409,
|
||
'UPLOAD_SESSION_CANCELLED',
|
||
)
|
||
}
|
||
|
||
async function writeStoredSeal(
|
||
uploadId: string,
|
||
seal: StoredUploadCompletionSeal,
|
||
): Promise<void> {
|
||
const sealPath = await resolveSealPath(uploadId, true)
|
||
if (!sealPath) {
|
||
throw new MaterialUploadRequestError(
|
||
'无法创建上传完成标记，已停止写入',
|
||
503,
|
||
'UPLOAD_SESSION_SEAL_UNAVAILABLE',
|
||
)
|
||
}
|
||
|
||
const partialPath = `${sealPath}.partial-${randomUUID()}`
|
||
let fileHandle: Awaited<ReturnType<typeof open>> | null = null
|
||
try {
|
||
fileHandle = await open(partialPath, 'wx', 0o600)
|
||
await fileHandle.writeFile(JSON.stringify(seal), 'utf8')
|
||
await fileHandle.sync()
|
||
await fileHandle.close()
|
||
fileHandle = null
|
||
|
||
// 生产 Linux/NAS 上 rename 会原子替换旧标记，不能先 rm 留出“无标记”
|
||
// 窗口；否则进程恰在两步之间退出时，迟到分片可能重新发布。Windows 仅
|
||
// 用于本地/Jest，且由同进程读写锁保护，需要先移除目标以兼容其语义。
|
||
if (process.platform === 'win32') {
|
||
await rm(sealPath, { force: true })
|
||
}
|
||
await rename(partialPath, sealPath)
|
||
} catch (error) {
|
||
if (fileHandle) {
|
||
await fileHandle.close().catch(() => undefined)
|
||
}
|
||
await rm(partialPath, { force: true }).catch(() => undefined)
|
||
if (error instanceof MaterialUploadRequestError) {
|
||
throw error
|
||
}
|
||
throw new MaterialUploadRequestError(
|
||
'无法持久化上传完成标记，已停止写入',
|
||
503,
|
||
'UPLOAD_SESSION_SEAL_UNAVAILABLE',
|
||
)
|
||
}
|
||
}
|