876 lines
30 KiB
TypeScript
876 lines
30 KiB
TypeScript
import type { Prisma } from '@prisma/client'
|
||
import { randomUUID } from 'crypto'
|
||
import { prisma } from '@/lib/prisma'
|
||
import {
|
||
assertOwnedProductPackageUploadSessions,
|
||
assertTaskDetailUploadAssetLimits,
|
||
completeProductPackageUpload,
|
||
type ProductPackageAssetsInput,
|
||
type CompleteProductPackageUploadInput,
|
||
} from '@/lib/product-package-completion'
|
||
import { refreshOwnedUploadSessionTtls } from '@/lib/upload-sessions'
|
||
import { TASK_DETAIL_UPLOAD_LIMITS } from '@/lib/task-detail-upload-policy'
|
||
import {
|
||
acquireTaskDetailUploadExecutionLock,
|
||
ensureTaskDetailUploadDrainCapability,
|
||
getTaskDetailUploadWorkerIdentity,
|
||
isTaskDetailUploadDraining,
|
||
removeTaskDetailUploadActiveMarker,
|
||
writeTaskDetailUploadActiveMarker,
|
||
} from '@/lib/task-detail-upload-runtime'
|
||
|
||
export const TASK_DETAIL_UPLOAD_JOB_STATUS = {
|
||
QUEUED: 'queued',
|
||
PROCESSING: 'processing',
|
||
COMPLETED: 'completed',
|
||
FAILED: 'failed',
|
||
} as const
|
||
|
||
export const TASK_DETAIL_UPLOAD_JOB_STAGE = {
|
||
QUEUED: 'queued',
|
||
VALIDATING: 'validating',
|
||
ASSEMBLING: 'assembling',
|
||
PREVIEWS: 'previews',
|
||
ZIPS: 'zips',
|
||
DATABASE: 'database',
|
||
CLEANUP: 'cleanup',
|
||
COMPLETED: 'completed',
|
||
FAILED: 'failed',
|
||
} as const
|
||
|
||
export type TaskDetailUploadJobStatus =
|
||
(typeof TASK_DETAIL_UPLOAD_JOB_STATUS)[keyof typeof TASK_DETAIL_UPLOAD_JOB_STATUS]
|
||
|
||
export interface CreateTaskDetailUploadJobInput {
|
||
materialId: string
|
||
taskNo?: string | null
|
||
dingtalkApprovalNo?: string | null
|
||
operatorUserId: string
|
||
assets: ProductPackageAssetsInput
|
||
options?: CompleteProductPackageUploadInput['options']
|
||
}
|
||
|
||
export interface TaskDetailUploadJobStatusView {
|
||
jobId: string
|
||
materialId: string
|
||
status: TaskDetailUploadJobStatus | string
|
||
stage: string
|
||
progress: number
|
||
error: string | null
|
||
createdAt: string
|
||
startedAt: string | null
|
||
completedAt: string | null
|
||
}
|
||
|
||
export class TaskDetailUploadQueueCapacityError extends Error {
|
||
readonly scope: 'user' | 'global'
|
||
readonly statusCode: 429 | 503
|
||
|
||
constructor(scope: 'user' | 'global') {
|
||
super(scope === 'user'
|
||
? `您已有 ${TASK_DETAIL_UPLOAD_LIMITS.maxQueuedJobsPerUser} 个任务等待处理，请稍后再提交`
|
||
: '当前上传处理队列已满，请稍后再提交')
|
||
this.name = 'TaskDetailUploadQueueCapacityError'
|
||
this.scope = scope
|
||
this.statusCode = scope === 'user' ? 429 : 503
|
||
}
|
||
}
|
||
|
||
const MAX_CONCURRENT_JOBS = 1
|
||
const DEFAULT_JOB_LEASE_MS = 5 * 60 * 1000
|
||
const DEFAULT_JOB_HEARTBEAT_MS = 30 * 1000
|
||
// 仅用于首次升级前没有 leaseToken 的旧 processing 任务。新任务只看数据库租约。
|
||
const DEFAULT_LEGACY_STALE_JOB_MS = 30 * 60 * 1000
|
||
const RECOVERY_BATCH_SIZE = 20
|
||
const WORKER_ADVISORY_LOCK_NAME = 'product-package-upload-worker'
|
||
|
||
let ensureTablePromise: Promise<void> | null = null
|
||
const pendingJobIds = new Set<string>()
|
||
let activeJobCount = 0
|
||
let pumpRunning = false
|
||
let globallyBackpressured = false
|
||
let retryTimer: ReturnType<typeof setTimeout> | null = null
|
||
const GLOBAL_RETRY_DELAY_MS = 5_000
|
||
|
||
/**
|
||
* 在当前事务内获取任务-详情上传的 advisory lock。
|
||
*
|
||
* pg_advisory_xact_lock() 本身返回 PostgreSQL 的 void 类型。直接用
|
||
* Prisma $queryRaw 读取它会触发 `Failed to deserialize column of type 'void'`，
|
||
* 因此把函数放进子查询，只返回一个普通整数列给 Prisma；锁仍会在当前
|
||
* 事务结束时自动释放。
|
||
*/
|
||
export async function acquireTaskDetailUploadLock(
|
||
tx: Prisma.TransactionClient,
|
||
lockName: string,
|
||
) {
|
||
await tx.$queryRaw`
|
||
SELECT 1 AS locked
|
||
FROM (
|
||
SELECT pg_advisory_xact_lock(hashtext(${lockName}))
|
||
) AS lock_result
|
||
`
|
||
}
|
||
|
||
function getLegacyStaleJobMs() {
|
||
const value = Number(process.env.TASK_DETAIL_UPLOAD_STALE_MS)
|
||
return Number.isFinite(value) && value > 0 ? value : DEFAULT_LEGACY_STALE_JOB_MS
|
||
}
|
||
|
||
function getJobLeaseMs() {
|
||
const value = Number(process.env.TASK_DETAIL_UPLOAD_LEASE_MS)
|
||
return Number.isFinite(value) && value >= 60_000 ? value : DEFAULT_JOB_LEASE_MS
|
||
}
|
||
|
||
function getJobHeartbeatMs() {
|
||
const value = Number(process.env.TASK_DETAIL_UPLOAD_HEARTBEAT_MS)
|
||
return Number.isFinite(value) && value >= 1_000 ? value : DEFAULT_JOB_HEARTBEAT_MS
|
||
}
|
||
|
||
async function readDatabaseClock(client: any): Promise<Date> {
|
||
const rows = await client.$queryRawUnsafe('SELECT clock_timestamp() AS "now"')
|
||
const value = rows?.[0]?.now
|
||
const now = value instanceof Date ? value : new Date(value)
|
||
if (!value || Number.isNaN(now.getTime())) {
|
||
throw new Error('无法读取数据库当前时间，拒绝执行任务详情作业租约操作')
|
||
}
|
||
return now
|
||
}
|
||
|
||
export class TaskDetailUploadLeaseLostError extends Error {
|
||
constructor() {
|
||
super('任务详情后台作业租约已失效，旧进程结果已拒绝')
|
||
this.name = 'TaskDetailUploadLeaseLostError'
|
||
}
|
||
}
|
||
|
||
function toJsonValue(value: unknown): Prisma.InputJsonValue {
|
||
return JSON.parse(JSON.stringify(value)) as Prisma.InputJsonValue
|
||
}
|
||
|
||
function toIso(value: Date | null | undefined) {
|
||
return value ? value.toISOString() : null
|
||
}
|
||
|
||
function toStatusView(job: {
|
||
id: string
|
||
materialId: string
|
||
status: string
|
||
stage: string
|
||
progress: number
|
||
errorMessage: string | null
|
||
createdAt: Date
|
||
startedAt: Date | null
|
||
completedAt: Date | null
|
||
}): TaskDetailUploadJobStatusView {
|
||
return {
|
||
jobId: job.id,
|
||
materialId: job.materialId,
|
||
status: job.status,
|
||
stage: job.stage,
|
||
progress: Math.max(0, Math.min(100, job.progress)),
|
||
error: job.errorMessage || null,
|
||
createdAt: job.createdAt.toISOString(),
|
||
startedAt: toIso(job.startedAt),
|
||
completedAt: toIso(job.completedAt),
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 任务表由 reviewed Prisma migration 创建。请求路径不执行 DDL，避免滚动发布时
|
||
* 多个容器同时抢迁移锁；若迁移未执行，调用方会收到明确的数据库错误并阻止入队。
|
||
*/
|
||
export async function ensureProductPackageUploadJobTable() {
|
||
if (!ensureTablePromise) {
|
||
ensureTablePromise = Promise.resolve()
|
||
}
|
||
return ensureTablePromise
|
||
}
|
||
|
||
async function updateJobProgress(
|
||
jobId: string,
|
||
leaseToken: string,
|
||
update: { stage: string; progress: number; message?: string },
|
||
) {
|
||
const progress = Math.max(0, Math.min(99, Math.round(update.progress)))
|
||
const updated = await prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, WORKER_ADVISORY_LOCK_NAME)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
return tx.productPackageUploadJob.updateMany({
|
||
where: {
|
||
id: jobId,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
leaseToken,
|
||
leaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
data: {
|
||
stage: update.stage,
|
||
progress,
|
||
...(update.message ? { errorMessage: update.message } : {}),
|
||
},
|
||
})
|
||
})
|
||
if (updated.count !== 1) {
|
||
throw new TaskDetailUploadLeaseLostError()
|
||
}
|
||
}
|
||
|
||
async function claimTaskDetailUploadJob(jobId: string) {
|
||
const token = randomUUID()
|
||
const { workerId } = getTaskDetailUploadWorkerIdentity()
|
||
return prisma.$transaction(async (tx) => {
|
||
// 用事务级 advisory lock 串行化所有容器的“领取任务”动作，保证全局最多一个
|
||
// 任务进入 Sharp/ZIP 重处理，而不是只在单个 Node 进程内限流。
|
||
await acquireTaskDetailUploadLock(tx, WORKER_ADVISORY_LOCK_NAME)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
const legacyStaleBefore = new Date(databaseNow.getTime() - getLegacyStaleJobMs())
|
||
const activeCount = await tx.productPackageUploadJob.count({
|
||
where: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
OR: [
|
||
{ leaseToken: { not: null }, leaseExpiresAt: { gt: databaseNow } },
|
||
{ leaseToken: null, updatedAt: { gt: legacyStaleBefore } },
|
||
],
|
||
},
|
||
})
|
||
if (activeCount > 0) return { status: 'busy' as const }
|
||
|
||
const claimed = await tx.productPackageUploadJob.updateMany({
|
||
where: {
|
||
id: jobId,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
},
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.VALIDATING,
|
||
progress: 5,
|
||
startedAt: databaseNow,
|
||
completedAt: null,
|
||
errorMessage: null,
|
||
workerId,
|
||
leaseToken: token,
|
||
leaseExpiresAt: new Date(databaseNow.getTime() + getJobLeaseMs()),
|
||
},
|
||
})
|
||
return claimed.count === 1
|
||
? { status: 'acquired' as const, token }
|
||
: { status: 'skip' as const }
|
||
})
|
||
}
|
||
|
||
async function isTaskDetailUploadLeaseCurrent(jobId: string, leaseToken: string) {
|
||
return prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, WORKER_ADVISORY_LOCK_NAME)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
return (await tx.productPackageUploadJob.count({
|
||
where: {
|
||
id: jobId,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
leaseToken,
|
||
leaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
})) === 1
|
||
})
|
||
}
|
||
|
||
async function renewTaskDetailUploadLease(jobId: string, leaseToken: string) {
|
||
const renewed = await prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, WORKER_ADVISORY_LOCK_NAME)
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
return tx.productPackageUploadJob.updateMany({
|
||
where: {
|
||
id: jobId,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
leaseToken,
|
||
leaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
data: {
|
||
leaseExpiresAt: new Date(databaseNow.getTime() + getJobLeaseMs()),
|
||
},
|
||
})
|
||
})
|
||
return renewed.count === 1
|
||
}
|
||
|
||
async function releaseTaskDetailUploadLease(
|
||
jobId: string,
|
||
leaseToken: string,
|
||
resetToQueued = false,
|
||
) {
|
||
return prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, WORKER_ADVISORY_LOCK_NAME)
|
||
const released = await tx.productPackageUploadJob.updateMany({
|
||
where: { id: jobId, leaseToken },
|
||
data: {
|
||
workerId: null,
|
||
leaseToken: null,
|
||
leaseExpiresAt: null,
|
||
...(resetToQueued ? {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.QUEUED,
|
||
progress: 0,
|
||
startedAt: null,
|
||
completedAt: null,
|
||
} : {}),
|
||
},
|
||
})
|
||
return released.count === 1
|
||
})
|
||
}
|
||
|
||
async function reconcileCommittedTaskDetailUploadJob(jobId: string) {
|
||
const job = await prisma.productPackageUploadJob.findUnique({
|
||
where: { id: jobId },
|
||
select: {
|
||
id: true,
|
||
status: true,
|
||
materialId: true,
|
||
material: { select: { zipFilePath: true, taskStatus: true } },
|
||
},
|
||
})
|
||
if (!job) return true
|
||
if (job.status !== TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED && job.status !== TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING) {
|
||
return true
|
||
}
|
||
|
||
const assetCount = await prisma.materialAsset.count({ where: { materialId: job.materialId } })
|
||
if (job.material?.taskStatus !== 'completed' || !job.material.zipFilePath || assetCount === 0) return false
|
||
|
||
// 主资产事务已提交但容器可能在 job.completed 写入前被杀掉：不要拿已清理的
|
||
// upload session 重跑，直接把持久化任务补记为完成。
|
||
await prisma.productPackageUploadJob.updateMany({
|
||
where: { id: jobId, status: { in: [TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED, TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING] } },
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.COMPLETED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.COMPLETED,
|
||
progress: 100,
|
||
completedAt: new Date(),
|
||
errorMessage: null,
|
||
workerId: null,
|
||
leaseToken: null,
|
||
leaseExpiresAt: null,
|
||
},
|
||
})
|
||
return true
|
||
}
|
||
|
||
type TaskDetailUploadProcessOutcome = 'processed' | 'busy' | 'ignored'
|
||
|
||
async function processOneTaskDetailUploadJob(jobId: string): Promise<TaskDetailUploadProcessOutcome> {
|
||
await ensureTaskDetailUploadDrainCapability()
|
||
if (await isTaskDetailUploadDraining()) return 'busy'
|
||
if (await reconcileCommittedTaskDetailUploadJob(jobId)) return 'ignored'
|
||
let releaseExecutionLock: Awaited<ReturnType<typeof acquireTaskDetailUploadExecutionLock>>
|
||
try {
|
||
releaseExecutionLock = await acquireTaskDetailUploadExecutionLock()
|
||
} catch (error) {
|
||
// 获取不到共享 uploads 上的 flock 时必须保守背压，不允许降级成容器内并发。
|
||
console.error(`[TaskDetailUploadJob] execution lock unavailable job=${jobId}:`, error)
|
||
return 'busy'
|
||
}
|
||
|
||
try {
|
||
// 数据库租约只决定谁能写结果；独立 flock 覆盖从领取到清理的整个真实执行期，
|
||
// 避免旧进程心跳超时但 Sharp/ZIP 仍未停止时，新进程再开一套重处理。
|
||
releaseExecutionLock.assertHeld()
|
||
const claim = await claimTaskDetailUploadJob(jobId)
|
||
if (claim.status !== 'acquired') {
|
||
const pendingJob = await prisma.productPackageUploadJob.findUnique({
|
||
where: { id: jobId },
|
||
select: { status: true },
|
||
})
|
||
if (pendingJob?.status === TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED) {
|
||
return 'busy'
|
||
}
|
||
return 'ignored'
|
||
}
|
||
const leaseToken = claim.token
|
||
|
||
const job = await prisma.productPackageUploadJob.findUnique({ where: { id: jobId } })
|
||
if (!job) {
|
||
await releaseTaskDetailUploadLease(jobId, leaseToken).catch(() => undefined)
|
||
return 'ignored'
|
||
}
|
||
|
||
let leaseLost = false
|
||
let resetToQueuedAfterDrain = false
|
||
let heartbeatTimer: ReturnType<typeof setTimeout> | null = null
|
||
let heartbeatInFlight: Promise<void> = Promise.resolve()
|
||
let heartbeatStopped = false
|
||
const abortController = new AbortController()
|
||
const markExecutionLost = (error: unknown) => {
|
||
leaseLost = true
|
||
heartbeatStopped = true
|
||
if (!abortController.signal.aborted) {
|
||
abortController.abort(error instanceof Error ? error : new Error(String(error)))
|
||
}
|
||
}
|
||
const assertExecutionCurrent = async () => {
|
||
if (leaseLost || abortController.signal.aborted) {
|
||
throw new TaskDetailUploadLeaseLostError()
|
||
}
|
||
try {
|
||
releaseExecutionLock.assertHeld()
|
||
} catch (error) {
|
||
markExecutionLost(error)
|
||
throw new TaskDetailUploadLeaseLostError()
|
||
}
|
||
if (!(await isTaskDetailUploadLeaseCurrent(jobId, leaseToken))) {
|
||
const error = new TaskDetailUploadLeaseLostError()
|
||
markExecutionLost(error)
|
||
throw error
|
||
}
|
||
}
|
||
const scheduleHeartbeat = () => {
|
||
heartbeatTimer = setTimeout(() => {
|
||
heartbeatTimer = null
|
||
heartbeatInFlight = (async () => {
|
||
try {
|
||
releaseExecutionLock.assertHeld()
|
||
if (!(await renewTaskDetailUploadLease(jobId, leaseToken))) {
|
||
throw new TaskDetailUploadLeaseLostError()
|
||
}
|
||
await writeTaskDetailUploadActiveMarker(jobId, leaseToken)
|
||
} catch (heartbeatError) {
|
||
markExecutionLost(heartbeatError)
|
||
console.error(`[TaskDetailUploadJob] lease heartbeat failed job=${jobId}:`, heartbeatError)
|
||
}
|
||
})().finally(() => {
|
||
if (!heartbeatStopped) scheduleHeartbeat()
|
||
})
|
||
}, getJobHeartbeatMs())
|
||
;(heartbeatTimer as NodeJS.Timeout).unref?.()
|
||
}
|
||
const stopHeartbeat = async () => {
|
||
heartbeatStopped = true
|
||
if (heartbeatTimer) {
|
||
clearTimeout(heartbeatTimer)
|
||
heartbeatTimer = null
|
||
}
|
||
await heartbeatInFlight.catch(() => undefined)
|
||
}
|
||
|
||
try {
|
||
// active 标记必须先落盘。部署 drain 可能出现在领取租约与写 marker 之间，
|
||
// 所以写完后再次检查；命中就释放租约并保持 queued，绝不开始图片/ZIP处理。
|
||
await writeTaskDetailUploadActiveMarker(jobId, leaseToken)
|
||
if (await isTaskDetailUploadDraining()) {
|
||
resetToQueuedAfterDrain = true
|
||
return 'busy'
|
||
}
|
||
scheduleHeartbeat()
|
||
|
||
try {
|
||
await completeProductPackageUpload({
|
||
materialId: job.materialId,
|
||
taskNo: job.taskNo,
|
||
dingtalkApprovalNo: job.dingtalkApprovalNo,
|
||
operatorUserId: job.operatorUserId,
|
||
assets: job.assets as ProductPackageAssetsInput,
|
||
options: {
|
||
...(job.options && typeof job.options === 'object' ? job.options as Record<string, unknown> : {}),
|
||
preserveMaterialOnError: true,
|
||
signal: abortController.signal,
|
||
assertExecutionCurrent,
|
||
onProgress: (update) => updateJobProgress(jobId, leaseToken, update),
|
||
},
|
||
})
|
||
await assertExecutionCurrent()
|
||
} catch (error) {
|
||
await stopHeartbeat()
|
||
if (leaseLost || error instanceof TaskDetailUploadLeaseLostError) {
|
||
console.warn(`[TaskDetailUploadJob] stale worker stopped without writing result job=${jobId}`)
|
||
return 'processed'
|
||
}
|
||
try {
|
||
await assertExecutionCurrent()
|
||
} catch {
|
||
console.warn(`[TaskDetailUploadJob] stale worker stopped without writing failure job=${jobId}`)
|
||
return 'processed'
|
||
}
|
||
const errorMessage = error instanceof Error ? error.message : '任务-详情上传处理失败'
|
||
console.error(`[TaskDetailUploadJob] failed job=${jobId}:`, error)
|
||
try {
|
||
await prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, WORKER_ADVISORY_LOCK_NAME)
|
||
releaseExecutionLock.assertHeld()
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
const failed = await tx.productPackageUploadJob.updateMany({
|
||
where: {
|
||
id: jobId,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
leaseToken,
|
||
leaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.FAILED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.FAILED,
|
||
errorMessage: errorMessage.slice(0, 2000),
|
||
completedAt: databaseNow,
|
||
workerId: null,
|
||
leaseToken: null,
|
||
leaseExpiresAt: null,
|
||
},
|
||
})
|
||
if (failed.count !== 1) throw new TaskDetailUploadLeaseLostError()
|
||
await tx.material.updateMany({
|
||
where: { id: job.materialId, taskStatus: { not: 'completed' } },
|
||
data: { taskStatus: 'failed' },
|
||
})
|
||
})
|
||
} catch (jobFailureRecordError) {
|
||
console.error(`[TaskDetailUploadJob] failed record write failed job=${jobId}:`, jobFailureRecordError)
|
||
}
|
||
return 'processed'
|
||
}
|
||
|
||
await stopHeartbeat()
|
||
try {
|
||
await assertExecutionCurrent()
|
||
} catch (error) {
|
||
console.warn(`[TaskDetailUploadJob] stale worker stopped before completed write job=${jobId}:`, error)
|
||
return 'processed'
|
||
}
|
||
try {
|
||
await prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, WORKER_ADVISORY_LOCK_NAME)
|
||
if (leaseLost || abortController.signal.aborted) throw new TaskDetailUploadLeaseLostError()
|
||
releaseExecutionLock.assertHeld()
|
||
const databaseNow = await readDatabaseClock(tx)
|
||
if (leaseLost || abortController.signal.aborted) throw new TaskDetailUploadLeaseLostError()
|
||
const completed = await tx.productPackageUploadJob.updateMany({
|
||
where: {
|
||
id: jobId,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
leaseToken,
|
||
leaseExpiresAt: { gt: databaseNow },
|
||
},
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.COMPLETED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.COMPLETED,
|
||
progress: 100,
|
||
completedAt: databaseNow,
|
||
errorMessage: null,
|
||
workerId: null,
|
||
leaseToken: null,
|
||
leaseExpiresAt: null,
|
||
},
|
||
})
|
||
if (completed.count !== 1) throw new TaskDetailUploadLeaseLostError()
|
||
})
|
||
} catch (completionRecordError) {
|
||
// 素材处理已经成功时，不能把 job 反写成 failed。保留 processing，交给
|
||
// reconcile/周期恢复在数据库恢复后补记 completed。
|
||
console.error(`[TaskDetailUploadJob] completed record write failed job=${jobId}:`, completionRecordError)
|
||
if (completionRecordError instanceof TaskDetailUploadLeaseLostError) {
|
||
markExecutionLost(completionRecordError)
|
||
console.warn(`[TaskDetailUploadJob] stale worker skipped completed reconcile job=${jobId}`)
|
||
} else {
|
||
try {
|
||
await reconcileCommittedTaskDetailUploadJob(jobId)
|
||
} catch (reconcileError) {
|
||
console.error(`[TaskDetailUploadJob] completed record reconcile failed job=${jobId}:`, reconcileError)
|
||
}
|
||
}
|
||
}
|
||
} finally {
|
||
await stopHeartbeat()
|
||
await releaseTaskDetailUploadLease(
|
||
jobId,
|
||
leaseToken,
|
||
resetToQueuedAfterDrain || leaseLost,
|
||
).catch((error) => {
|
||
console.error(`[TaskDetailUploadJob] lease release failed job=${jobId}:`, error)
|
||
})
|
||
await removeTaskDetailUploadActiveMarker().catch((error) => {
|
||
console.error(`[TaskDetailUploadJob] active marker cleanup failed job=${jobId}:`, error)
|
||
})
|
||
}
|
||
return 'processed'
|
||
} finally {
|
||
await releaseExecutionLock().catch((error) => {
|
||
console.error(`[TaskDetailUploadJob] execution lock release failed job=${jobId}:`, error)
|
||
})
|
||
}
|
||
}
|
||
|
||
function scheduleGlobalQueueRetry() {
|
||
if (retryTimer) return
|
||
retryTimer = setTimeout(() => {
|
||
retryTimer = null
|
||
globallyBackpressured = false
|
||
void pumpTaskDetailUploadJobs()
|
||
}, GLOBAL_RETRY_DELAY_MS)
|
||
// 生产服务本身会常驻，但测试/CLI 不应被单个背压 timer 阻止退出。
|
||
;(retryTimer as ReturnType<typeof setTimeout> & { unref?: () => void }).unref?.()
|
||
}
|
||
|
||
async function pumpTaskDetailUploadJobs() {
|
||
if (pumpRunning || globallyBackpressured) {
|
||
return
|
||
}
|
||
|
||
pumpRunning = true
|
||
try {
|
||
await ensureTaskDetailUploadDrainCapability()
|
||
if (await isTaskDetailUploadDraining()) {
|
||
globallyBackpressured = true
|
||
scheduleGlobalQueueRetry()
|
||
return
|
||
}
|
||
while (activeJobCount < MAX_CONCURRENT_JOBS && pendingJobIds.size > 0) {
|
||
const nextJobId = pendingJobIds.values().next().value as string | undefined
|
||
if (!nextJobId) {
|
||
break
|
||
}
|
||
pendingJobIds.delete(nextJobId)
|
||
activeJobCount += 1
|
||
void processOneTaskDetailUploadJob(nextJobId)
|
||
.then((outcome) => {
|
||
if (outcome === 'busy') {
|
||
pendingJobIds.add(nextJobId)
|
||
globallyBackpressured = true
|
||
scheduleGlobalQueueRetry()
|
||
}
|
||
})
|
||
.catch((error) => {
|
||
console.error(`[TaskDetailUploadJob] worker crashed job=${nextJobId}:`, error)
|
||
pendingJobIds.add(nextJobId)
|
||
globallyBackpressured = true
|
||
scheduleGlobalQueueRetry()
|
||
})
|
||
.finally(() => {
|
||
activeJobCount -= 1
|
||
if (!globallyBackpressured) {
|
||
void pumpTaskDetailUploadJobs()
|
||
}
|
||
})
|
||
}
|
||
} finally {
|
||
pumpRunning = false
|
||
}
|
||
}
|
||
|
||
export function triggerTaskDetailUploadJob(jobId: string) {
|
||
pendingJobIds.add(jobId)
|
||
void pumpTaskDetailUploadJobs().catch((error) => {
|
||
console.error('[TaskDetailUploadJob] queue pump failed:', error)
|
||
})
|
||
}
|
||
|
||
export async function createTaskDetailUploadJob(input: CreateTaskDetailUploadJobInput) {
|
||
assertTaskDetailUploadAssetLimits(input.assets)
|
||
await assertOwnedProductPackageUploadSessions(input.assets, input.operatorUserId)
|
||
const uploadIds = [
|
||
...(input.assets.main || []),
|
||
...(input.assets.sku || []),
|
||
...(input.assets.detail || []),
|
||
...(input.assets.psd || []),
|
||
].flatMap((asset) => [
|
||
String(asset.uploadId || '').trim(),
|
||
...((asset as { previewAssets?: Array<{ uploadId?: string }> }).previewAssets || [])
|
||
.map((preview) => String(preview.uploadId || '').trim()),
|
||
]).filter(Boolean)
|
||
await refreshOwnedUploadSessionTtls(uploadIds)
|
||
await ensureProductPackageUploadJobTable()
|
||
const job = await prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, 'product-package-upload-create')
|
||
const existing = await tx.productPackageUploadJob.findFirst({
|
||
where: {
|
||
materialId: input.materialId,
|
||
operatorUserId: input.operatorUserId,
|
||
status: {
|
||
in: [
|
||
TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
TASK_DETAIL_UPLOAD_JOB_STATUS.COMPLETED,
|
||
],
|
||
},
|
||
},
|
||
orderBy: { createdAt: 'desc' },
|
||
})
|
||
if (existing) return existing
|
||
|
||
const queuedForUser = await tx.productPackageUploadJob.count({
|
||
where: {
|
||
operatorUserId: input.operatorUserId,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
},
|
||
})
|
||
if (queuedForUser >= TASK_DETAIL_UPLOAD_LIMITS.maxQueuedJobsPerUser) {
|
||
throw new TaskDetailUploadQueueCapacityError('user')
|
||
}
|
||
|
||
const queuedGlobally = await tx.productPackageUploadJob.count({
|
||
where: { status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED },
|
||
})
|
||
if (queuedGlobally >= TASK_DETAIL_UPLOAD_LIMITS.maxQueuedJobsGlobal) {
|
||
throw new TaskDetailUploadQueueCapacityError('global')
|
||
}
|
||
|
||
const created = await tx.productPackageUploadJob.create({
|
||
data: {
|
||
materialId: input.materialId,
|
||
operatorUserId: input.operatorUserId,
|
||
taskNo: input.taskNo?.trim() || null,
|
||
dingtalkApprovalNo: input.dingtalkApprovalNo?.trim() || null,
|
||
assets: toJsonValue(input.assets),
|
||
options: input.options ? toJsonValue(input.options) : undefined,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.QUEUED,
|
||
progress: 0,
|
||
errorMessage: null,
|
||
},
|
||
})
|
||
|
||
// 素材列表/详情页可以据此识别“正在后台处理”的空素材，用户关闭上传页后
|
||
// 仍然能知道它不是一个普通的空卡片。完成函数会把状态推进到 completed。
|
||
await tx.material.update({
|
||
where: { id: input.materialId },
|
||
data: { taskStatus: 'uploading' },
|
||
})
|
||
|
||
return created
|
||
})
|
||
|
||
if (job.status === TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED) {
|
||
triggerTaskDetailUploadJob(job.id)
|
||
}
|
||
return toStatusView(job)
|
||
}
|
||
|
||
export async function getTaskDetailUploadJobStatus(jobId: string, operatorUserId: string) {
|
||
await ensureProductPackageUploadJobTable()
|
||
const job = await prisma.productPackageUploadJob.findFirst({
|
||
where: {
|
||
id: jobId,
|
||
operatorUserId,
|
||
},
|
||
})
|
||
|
||
return job ? toStatusView(job) : null
|
||
}
|
||
|
||
export async function getLatestTaskDetailUploadJobStatus(materialId: string, operatorUserId: string) {
|
||
await ensureProductPackageUploadJobTable()
|
||
const job = await prisma.productPackageUploadJob.findFirst({
|
||
where: {
|
||
materialId,
|
||
operatorUserId,
|
||
},
|
||
orderBy: { createdAt: 'desc' },
|
||
})
|
||
|
||
return job ? toStatusView(job) : null
|
||
}
|
||
|
||
/**
|
||
* 服务重启后恢复未完成队列。processing 会在心跳停止并超过 stale 窗口后标记失败，
|
||
* 不直接重跑可能已经写入部分资产的非幂等处理链路。
|
||
*/
|
||
export async function recoverTaskDetailUploadJobs() {
|
||
await ensureProductPackageUploadJobTable()
|
||
await ensureTaskDetailUploadDrainCapability()
|
||
if (await isTaskDetailUploadDraining()) {
|
||
return 0
|
||
}
|
||
const databaseNow = await readDatabaseClock(prisma)
|
||
const staleBefore = new Date(databaseNow.getTime() - getLegacyStaleJobMs())
|
||
|
||
const staleJobs = await prisma.productPackageUploadJob.findMany({
|
||
where: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
OR: [
|
||
{ leaseToken: { not: null }, leaseExpiresAt: { lte: databaseNow } },
|
||
{ leaseToken: { not: null }, leaseExpiresAt: null },
|
||
{ leaseToken: null, updatedAt: { lt: staleBefore } },
|
||
],
|
||
},
|
||
select: { id: true, materialId: true, leaseToken: true },
|
||
})
|
||
|
||
// 完成链路现在以素材锁、会话锁和单事务提交保证幂等。过期租约只回到 queued，
|
||
// 源上传会话和素材都保留；新 worker 会重新验证并安全收口，而不是要求用户重传。
|
||
for (const staleJob of staleJobs) {
|
||
if (await reconcileCommittedTaskDetailUploadJob(staleJob.id)) continue
|
||
|
||
await prisma.$transaction(async (tx) => {
|
||
await acquireTaskDetailUploadLock(tx, WORKER_ADVISORY_LOCK_NAME)
|
||
const lockedNow = await readDatabaseClock(tx)
|
||
const lockedLegacyStaleBefore = new Date(lockedNow.getTime() - getLegacyStaleJobMs())
|
||
const released = await tx.productPackageUploadJob.updateMany({
|
||
where: {
|
||
id: staleJob.id,
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.PROCESSING,
|
||
...(staleJob.leaseToken
|
||
? {
|
||
leaseToken: staleJob.leaseToken,
|
||
OR: [
|
||
{ leaseExpiresAt: { lte: lockedNow } },
|
||
{ leaseExpiresAt: null },
|
||
],
|
||
}
|
||
: { leaseToken: null, updatedAt: { lt: lockedLegacyStaleBefore } }),
|
||
},
|
||
data: {
|
||
status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED,
|
||
stage: TASK_DETAIL_UPLOAD_JOB_STAGE.QUEUED,
|
||
progress: 0,
|
||
errorMessage: '后台处理进程中断，系统已自动恢复排队',
|
||
startedAt: null,
|
||
completedAt: null,
|
||
workerId: null,
|
||
leaseToken: null,
|
||
leaseExpiresAt: null,
|
||
},
|
||
})
|
||
if (released.count === 1) {
|
||
await tx.material.updateMany({
|
||
where: { id: staleJob.materialId, taskStatus: { not: 'completed' } },
|
||
data: { taskStatus: 'uploading' },
|
||
})
|
||
}
|
||
})
|
||
}
|
||
|
||
const queuedJobs = await prisma.productPackageUploadJob.findMany({
|
||
where: { status: TASK_DETAIL_UPLOAD_JOB_STATUS.QUEUED },
|
||
orderBy: { createdAt: 'asc' },
|
||
take: RECOVERY_BATCH_SIZE,
|
||
select: { id: true },
|
||
})
|
||
|
||
for (const job of queuedJobs) {
|
||
triggerTaskDetailUploadJob(job.id)
|
||
}
|
||
|
||
if (queuedJobs.length > 0) {
|
||
console.log(`[TaskDetailUploadJob] recovered ${queuedJobs.length} queued job(s)`)
|
||
}
|
||
return queuedJobs.length
|
||
}
|
||
|
||
/** @internal 仅供队列核心单元测试使用，不由任何 HTTP 路由调用。 */
|
||
export const __taskDetailUploadJobTesting = {
|
||
claimTaskDetailUploadJob,
|
||
reconcileCommittedTaskDetailUploadJob,
|
||
processOneTaskDetailUploadJob,
|
||
pumpTaskDetailUploadJobs,
|
||
getRuntimeState: () => ({
|
||
pendingJobIds: Array.from(pendingJobIds),
|
||
activeJobCount,
|
||
pumpRunning,
|
||
globallyBackpressured,
|
||
retryScheduled: retryTimer !== null,
|
||
}),
|
||
resetRuntimeState: () => {
|
||
if (retryTimer) clearTimeout(retryTimer)
|
||
retryTimer = null
|
||
globallyBackpressured = false
|
||
pendingJobIds.clear()
|
||
activeJobCount = 0
|
||
pumpRunning = false
|
||
ensureTablePromise = null
|
||
},
|
||
}
|