merge: release bounded BOM worker execution
This commit is contained in:
WuYongJie
commit
3cbbe8eb3d
@ -15,7 +15,7 @@ jest.mock('@/server/services/bom-costing/bom-costing-worker.service', () => ({
|
||||
},
|
||||
}))
|
||||
|
||||
const { POST } = require('./route')
|
||||
const { POST, maxDuration } = require('./route')
|
||||
|
||||
function createRequest(secret?: string) {
|
||||
return {
|
||||
@ -46,6 +46,10 @@ describe('POST /api/internal/bom-costing/worker/run', () => {
|
||||
process.env.BOM_COSTING_WORKER_SECRET = originalSecret
|
||||
})
|
||||
|
||||
it('将单轮后台处理限制在二十分钟以内', () => {
|
||||
expect(maxDuration).toBe(1200)
|
||||
})
|
||||
|
||||
it('拒绝没有内部任务密钥的请求', async () => {
|
||||
const response = await POST(createRequest() as never)
|
||||
|
||||
@ -58,7 +62,9 @@ describe('POST /api/internal/bom-costing/worker/run', () => {
|
||||
|
||||
const response = await POST(createRequest('worker-test-secret') as never)
|
||||
|
||||
expect(mockRunNext).toHaveBeenCalledTimes(1)
|
||||
expect(mockRunNext).toHaveBeenCalledWith({
|
||||
signal: expect.any(AbortSignal),
|
||||
})
|
||||
expect(response).toEqual({
|
||||
success: true,
|
||||
data: { worker: { status: 'idle' } },
|
||||
|
||||
@ -1,15 +1,27 @@
|
||||
import { timingSafeEqual } from 'node:crypto'
|
||||
import { NextRequest } from 'next/server'
|
||||
import { ErrorCode } from '@/server/core/errors'
|
||||
import { AppError, ErrorCode } from '@/server/core/errors'
|
||||
import { jsonError, jsonSuccess } from '@/server/core/response'
|
||||
import { bomCostingWorkerService } from '@/server/services/bom-costing/bom-costing-worker.service'
|
||||
|
||||
export const dynamic = 'force-dynamic'
|
||||
// A single worker turn may synthesize a large SKU catalog after the image
|
||||
// facts have been reused. Keep this bounded, but do not cut a healthy run off
|
||||
// at the old 30-minute HTTP client limit. This is still a finite safety cap;
|
||||
// stalled batches are split by the analyzer instead of retrying indefinitely.
|
||||
export const maxDuration = 2700
|
||||
// A worker turn has bounded image/model calls and must never wait for hours.
|
||||
// The deployment script applies the same 20-minute ceiling to its curl loop.
|
||||
export const maxDuration = 1200
|
||||
|
||||
function createWorkerTurnDeadline() {
|
||||
const controller = new AbortController()
|
||||
const timeoutId = setTimeout(() => {
|
||||
controller.abort(new AppError(
|
||||
ErrorCode.BOM_COSTING_INVALID_PROVIDER,
|
||||
'BOM 核算单次处理已超过 20 分钟，任务已停止以避免长时间占用；已保存进度可在稍后继续使用',
|
||||
))
|
||||
}, maxDuration * 1_000)
|
||||
return {
|
||||
signal: controller.signal,
|
||||
cancel: () => clearTimeout(timeoutId),
|
||||
}
|
||||
}
|
||||
|
||||
function hasValidWorkerSecret(request: NextRequest) {
|
||||
const expected = process.env.BOM_COSTING_WORKER_SECRET?.trim()
|
||||
@ -33,8 +45,11 @@ export async function POST(request: NextRequest) {
|
||||
)
|
||||
}
|
||||
|
||||
const deadline = createWorkerTurnDeadline()
|
||||
try {
|
||||
return jsonSuccess({ worker: await bomCostingWorkerService.runNext() })
|
||||
return jsonSuccess({
|
||||
worker: await bomCostingWorkerService.runNext({ signal: deadline.signal }),
|
||||
})
|
||||
} catch (error) {
|
||||
console.error('[BOM costing worker] task execution failed', error)
|
||||
return jsonError(
|
||||
@ -42,5 +57,7 @@ export async function POST(request: NextRequest) {
|
||||
'BOM 核算后台任务执行失败',
|
||||
500,
|
||||
)
|
||||
} finally {
|
||||
deadline.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
@ -787,6 +787,12 @@ function getHistoricalOvertimeStages(item: OaBoardProjectListItemDTO): OvertimeS
|
||||
function resolveTimeoutFollowupOwner(item: OaBoardProjectListItemDTO, stageKey: string) {
|
||||
const stage = getOvertimeStages(item).find((entry) => entry.stageKey === stageKey)
|
||||
if (stage?.ownerLabel?.trim()) return stage.ownerLabel.trim()
|
||||
|
||||
const isHistoricalStage = getHistoricalOvertimeStages(item).some(
|
||||
(entry) => entry.stageKey === stageKey,
|
||||
)
|
||||
if (isHistoricalStage) return ''
|
||||
|
||||
const resolvedCurrentOwner = resolveCurrentOwnerDisplay(item)
|
||||
return resolvedCurrentOwner && resolvedCurrentOwner !== '-' ? resolvedCurrentOwner : '待分配'
|
||||
}
|
||||
@ -1671,7 +1677,7 @@ function TimeoutFollowupInspector({
|
||||
}))
|
||||
setDirtyStageKeys((current) => new Set(current).add(selectedStage.stageKey))
|
||||
}
|
||||
const dirtyEntries = currentOvertimeStages
|
||||
const dirtyEntries = overtimeStages
|
||||
.filter((stage) => dirtyStageKeys.has(stage.stageKey))
|
||||
.map((stage) => ({
|
||||
stage,
|
||||
@ -1683,7 +1689,6 @@ function TimeoutFollowupInspector({
|
||||
),
|
||||
}))
|
||||
const handleSave = async () => {
|
||||
if (isHistoricalStage) return
|
||||
const entries =
|
||||
dirtyEntries.length > 0 ? dirtyEntries : [{ stage: selectedStage, draft }]
|
||||
const saved = await onSave(entries)
|
||||
@ -1746,7 +1751,13 @@ function TimeoutFollowupInspector({
|
||||
>
|
||||
<span className="text-xs font-apple-semibold">{stage.stageLabel}</span>
|
||||
<span className={`mt-1 text-[10px] ${active ? 'text-white/70' : 'text-apple-label-secondary'}`}>
|
||||
{isHistorical ? '历史记录' : isDirty ? '待保存' : record ? getTimeoutFollowupStatusLabel(record.status) : '待跟进'}
|
||||
{isDirty
|
||||
? '待保存'
|
||||
: record
|
||||
? getTimeoutFollowupStatusLabel(record.status)
|
||||
: isHistorical
|
||||
? '未填写跟进'
|
||||
: '待跟进'}
|
||||
</span>
|
||||
{deadlineStatus ? (
|
||||
<span className={`mt-1 text-[10px] font-apple-semibold ${active ? 'text-rose-200' : 'text-rose-700'}`}>
|
||||
@ -1776,46 +1787,11 @@ function TimeoutFollowupInspector({
|
||||
</section>
|
||||
|
||||
{isHistoricalStage ? (
|
||||
<section className="mt-6 border border-apple-separator bg-[#fafbfc] p-4">
|
||||
<p className="text-xs text-apple-label-secondary">
|
||||
该阶段已离开当前节点，因此只作为历史超时保留，不会再触发钉钉待办。
|
||||
{selectedFollowup ? ' 以下为已保存的历史跟进内容。' : ' 该阶段尚未填写历史跟进记录。'}
|
||||
</p>
|
||||
<dl className="mt-4 grid gap-4 sm:grid-cols-2">
|
||||
<div>
|
||||
<dt className="text-[11px] font-apple-medium text-apple-label-secondary">跟进状态</dt>
|
||||
<dd className="mt-1"><StatusBadge label={getTimeoutFollowupStatusLabel(selectedFollowup?.status ?? 'PENDING')} tone="default" /></dd>
|
||||
</div>
|
||||
<div>
|
||||
<dt className="text-[11px] font-apple-medium text-apple-label-secondary">跟进负责人</dt>
|
||||
<dd className="mt-1 text-sm text-apple-label-primary">{formatCellText(selectedFollowup?.followUpOwner)}</dd>
|
||||
</div>
|
||||
<div>
|
||||
<dt className="text-[11px] font-apple-medium text-apple-label-secondary">预计完成日期</dt>
|
||||
<dd className={`mt-1 text-sm ${selectedFollowupDeadlineStatus ? 'font-apple-semibold text-rose-700' : 'text-apple-label-primary'}`}>
|
||||
{formatCellText(selectedFollowup?.expectedResolveDate)}
|
||||
{selectedFollowupDeadlineStatus ? (
|
||||
<span className="ml-2 text-xs">{getFollowupDeadlineLabel(selectedFollowupDeadlineStatus)}</span>
|
||||
) : null}
|
||||
</dd>
|
||||
</div>
|
||||
</dl>
|
||||
<div className="mt-5 border-t border-apple-separator pt-4">
|
||||
<p className="text-[11px] font-apple-medium text-apple-label-secondary">超时原因说明</p>
|
||||
<p className="mt-1 whitespace-pre-wrap text-sm leading-6 text-apple-label-primary">{formatCellText(selectedFollowup?.reasonDetail)}</p>
|
||||
</div>
|
||||
<div className="mt-5 border-t border-apple-separator pt-4">
|
||||
<p className="text-[11px] font-apple-medium text-apple-label-secondary">解决措施</p>
|
||||
<p className="mt-1 whitespace-pre-wrap text-sm leading-6 text-apple-label-primary">{formatCellText(selectedFollowup?.actionPlan)}</p>
|
||||
</div>
|
||||
{selectedFollowup?.updatedAt ? (
|
||||
<p className="mt-4 text-[11px] text-apple-label-secondary">
|
||||
最近更新：{formatCellText(selectedFollowup.updatedBy)} · {selectedFollowup.updatedAt.replace('T', ' ').slice(0, 16)}
|
||||
</p>
|
||||
) : null}
|
||||
</section>
|
||||
) : (
|
||||
<>
|
||||
<div className="mt-5 border border-amber-200 bg-amber-50 px-3 py-2 text-xs text-amber-800">
|
||||
历史超时跟进仅用于复盘，不会触发钉钉超时待办或计入当前超时。
|
||||
</div>
|
||||
) : null}
|
||||
<>
|
||||
<div className="mt-6 grid gap-4 sm:grid-cols-2">
|
||||
<FilterSelect
|
||||
label="跟进状态"
|
||||
@ -1876,8 +1852,7 @@ function TimeoutFollowupInspector({
|
||||
</p>
|
||||
) : null}
|
||||
{errorMessage ? <p className="mt-4 text-xs text-rose-700">{errorMessage}</p> : null}
|
||||
</>
|
||||
)}
|
||||
</>
|
||||
</div>
|
||||
|
||||
<footer className="flex shrink-0 items-center justify-end gap-3 border-t border-apple-separator px-6 py-4">
|
||||
@ -1888,17 +1863,15 @@ function TimeoutFollowupInspector({
|
||||
>
|
||||
关闭
|
||||
</button>
|
||||
{!isHistoricalStage ? (
|
||||
<button
|
||||
type="button"
|
||||
disabled={isSaving}
|
||||
onClick={() => void handleSave()}
|
||||
className={`inline-flex h-9 items-center gap-2 rounded-apple-md border px-4 text-sm font-apple-semibold transition ${WORKBENCH_PRIMARY_BUTTON_CLASS}`}
|
||||
>
|
||||
{isSaving ? <LoaderCircle className="h-4 w-4 animate-spin" /> : null}
|
||||
{dirtyEntries.length > 1 ? `保存 ${dirtyEntries.length} 个节点` : '保存跟进'}
|
||||
</button>
|
||||
) : null}
|
||||
<button
|
||||
type="button"
|
||||
disabled={isSaving}
|
||||
onClick={() => void handleSave()}
|
||||
className={`inline-flex h-9 items-center gap-2 rounded-apple-md border px-4 text-sm font-apple-semibold transition ${WORKBENCH_PRIMARY_BUTTON_CLASS}`}
|
||||
>
|
||||
{isSaving ? <LoaderCircle className="h-4 w-4 animate-spin" /> : null}
|
||||
{dirtyEntries.length > 1 ? `保存 ${dirtyEntries.length} 个节点` : '保存跟进'}
|
||||
</button>
|
||||
</footer>
|
||||
</section>
|
||||
</div>
|
||||
|
||||
@ -617,6 +617,7 @@ export async function extractBomCostingImageFacts(input: {
|
||||
snapshot: MarketDetailSnapshot
|
||||
existingSnapshot?: BomCostingSourceFactSnapshot | null
|
||||
analysisInstruction?: string
|
||||
signal?: AbortSignal
|
||||
onBatchStarted?: (progress: Omit<BomCostingFactBatchProgress, 'snapshot' | 'call'>) => Promise<void>
|
||||
onBatchCompleted?: (progress: BomCostingFactBatchProgress) => Promise<void>
|
||||
}) {
|
||||
@ -754,6 +755,7 @@ export async function extractBomCostingImageFacts(input: {
|
||||
userContent,
|
||||
maxTokens,
|
||||
timeoutMs,
|
||||
signal: input.signal,
|
||||
})
|
||||
let rawImages: unknown[]
|
||||
try {
|
||||
|
||||
@ -223,6 +223,8 @@ describe('BOM 多模态分析器', () => {
|
||||
process.env.BOM_COSTING_SYNTHESIS_SKU_BATCH_SIZE
|
||||
const originalSynthesisBatchConcurrency =
|
||||
process.env.BOM_COSTING_SYNTHESIS_BATCH_CONCURRENCY
|
||||
const originalSynthesisBatchMaxDuration =
|
||||
process.env.BOM_COSTING_SYNTHESIS_BATCH_MAX_DURATION_MS
|
||||
const originalMaxSkusPerCase = process.env.BOM_COSTING_MAX_SKUS_PER_CASE
|
||||
|
||||
beforeEach(() => {
|
||||
@ -263,6 +265,12 @@ describe('BOM 多模态分析器', () => {
|
||||
process.env.BOM_COSTING_SYNTHESIS_BATCH_CONCURRENCY =
|
||||
originalSynthesisBatchConcurrency
|
||||
}
|
||||
if (originalSynthesisBatchMaxDuration === undefined) {
|
||||
delete process.env.BOM_COSTING_SYNTHESIS_BATCH_MAX_DURATION_MS
|
||||
} else {
|
||||
process.env.BOM_COSTING_SYNTHESIS_BATCH_MAX_DURATION_MS =
|
||||
originalSynthesisBatchMaxDuration
|
||||
}
|
||||
if (originalMaxSkusPerCase === undefined) {
|
||||
delete process.env.BOM_COSTING_MAX_SKUS_PER_CASE
|
||||
} else {
|
||||
@ -1256,6 +1264,30 @@ describe('BOM 多模态分析器', () => {
|
||||
expect(output.result.skus).toHaveLength(2)
|
||||
})
|
||||
|
||||
it('最终汇总超过单批总时限时停止，不继续递归拆分占用后台任务', async () => {
|
||||
process.env.BOM_COSTING_SYNTHESIS_BATCH_MAX_DURATION_MS = '480000'
|
||||
let now = 0
|
||||
jest.spyOn(Date, 'now').mockImplementation(() => now)
|
||||
const timeoutError = new Error('The operation was aborted due to timeout')
|
||||
timeoutError.name = 'TimeoutError'
|
||||
const fetchSpy = jest.spyOn(global, 'fetch').mockImplementation(async () => {
|
||||
now = 480_000
|
||||
throw timeoutError
|
||||
})
|
||||
|
||||
await expect(
|
||||
new BomCostingMultimodalAnalyzer().analyzeWithTrace({
|
||||
sourceUrl: snapshot.detailUrl,
|
||||
snapshot,
|
||||
existingFactSnapshot: createCompleteFactSnapshot(snapshot),
|
||||
}),
|
||||
).rejects.toThrow(
|
||||
'第 1/1 批规格结构在 8 分钟内未完成，任务已停止以避免长时间占用',
|
||||
)
|
||||
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('模型网关暂时不可用时返回客户可读错误且不自动重试', async () => {
|
||||
const fetchSpy = jest.spyOn(global, 'fetch').mockResolvedValue(
|
||||
new Response(JSON.stringify({
|
||||
|
||||
@ -238,6 +238,11 @@ const DEFAULT_AUTOMATIC_SKU_SAFETY_LIMIT = 120
|
||||
// high-volume workflow may raise it up to the hard contract ceiling.
|
||||
const DEFAULT_SYNTHESIS_SKU_BATCH_SIZE = 8
|
||||
const DEFAULT_SYNTHESIS_BATCH_CONCURRENCY = 3
|
||||
// A synthesis batch is allowed to split a malformed or slow model response,
|
||||
// but it must still finish in a short, deterministic window. Without a wall
|
||||
// clock budget, one 4-SKU batch can recursively split and spend many minutes
|
||||
// retrying while the worker remains occupied.
|
||||
const DEFAULT_SYNTHESIS_BATCH_MAX_DURATION_MS = 8 * 60_000
|
||||
const MAX_EXPLICIT_FACT_SKU_IDS = 20
|
||||
const MAX_SYNTHESIS_CONTRACT_ATTEMPTS = 2
|
||||
const SYNTHESIS_IMAGE_REFERENCE_PREFIX = 'bom-image:'
|
||||
@ -621,6 +626,27 @@ function getSynthesisBatchConcurrency() {
|
||||
)
|
||||
}
|
||||
|
||||
function getSynthesisBatchMaxDurationMs() {
|
||||
return readBomCostingBoundedInteger(
|
||||
process.env.BOM_COSTING_SYNTHESIS_BATCH_MAX_DURATION_MS,
|
||||
DEFAULT_SYNTHESIS_BATCH_MAX_DURATION_MS,
|
||||
60_000,
|
||||
10 * 60_000,
|
||||
)
|
||||
}
|
||||
|
||||
function createSynthesisBatchDeadlineError(input: {
|
||||
batchIndex: number
|
||||
totalBatches: number
|
||||
maxDurationMs: number
|
||||
}) {
|
||||
const durationMinutes = Math.ceil(input.maxDurationMs / 60_000)
|
||||
return new AppError(
|
||||
ErrorCode.BOM_COSTING_INVALID_PROVIDER,
|
||||
`第 ${input.batchIndex}/${input.totalBatches} 批规格结构在 ${durationMinutes} 分钟内未完成，任务已停止以避免长时间占用；已保存的图片事实可在稍后继续使用`,
|
||||
)
|
||||
}
|
||||
|
||||
function hasSameSkuIds(left: string[], right: string[]) {
|
||||
return left.length === right.length && left.every((value, index) => (
|
||||
value === right[index]
|
||||
@ -867,6 +893,7 @@ export interface BomCostingMultimodalAnalyzeInput {
|
||||
existingFactSnapshot?: BomCostingSourceFactSnapshot | null
|
||||
synthesisCheckpoint?: BomCostingSynthesisCheckpoint | null
|
||||
analysisInstruction?: string
|
||||
signal?: AbortSignal
|
||||
onFactBatchStarted?: (
|
||||
progress: Omit<BomCostingFactBatchProgress, 'snapshot' | 'call'>,
|
||||
) => Promise<void>
|
||||
@ -923,6 +950,7 @@ export class BomCostingMultimodalAnalyzer {
|
||||
snapshot: input.snapshot,
|
||||
existingSnapshot: input.existingFactSnapshot,
|
||||
analysisInstruction: input.analysisInstruction,
|
||||
signal: input.signal,
|
||||
onBatchStarted: input.onFactBatchStarted,
|
||||
onBatchCompleted: input.onFactBatchCompleted,
|
||||
})
|
||||
@ -999,6 +1027,20 @@ export class BomCostingMultimodalAnalyzer {
|
||||
batchIndex: number,
|
||||
skuBatch: MarketSkuSnapshot[],
|
||||
) => {
|
||||
const synthesisBatchStartedAt = Date.now()
|
||||
const synthesisBatchMaxDurationMs = getSynthesisBatchMaxDurationMs()
|
||||
const getRemainingSynthesisBatchDurationMs = () => {
|
||||
const remainingMs =
|
||||
synthesisBatchMaxDurationMs - (Date.now() - synthesisBatchStartedAt)
|
||||
if (remainingMs > 0) return remainingMs
|
||||
throw createSynthesisBatchDeadlineError({
|
||||
batchIndex: batchIndex + 1,
|
||||
totalBatches: skuBatches.length,
|
||||
maxDurationMs: synthesisBatchMaxDurationMs,
|
||||
})
|
||||
}
|
||||
const hasReachedSynthesisBatchDeadline = () =>
|
||||
Date.now() - synthesisBatchStartedAt >= synthesisBatchMaxDurationMs
|
||||
await withCommitLock(async () => {
|
||||
await input.onSynthesisBatchStarted?.({
|
||||
batchIndex: batchIndex + 1,
|
||||
@ -1066,6 +1108,8 @@ export class BomCostingMultimodalAnalyzer {
|
||||
) {
|
||||
let responsePayload: BomCostingSynthesisCallResult
|
||||
try {
|
||||
const remainingSynthesisBatchDurationMs =
|
||||
getRemainingSynthesisBatchDurationMs()
|
||||
responsePayload = await callBomCostingMultimodalModel({
|
||||
config,
|
||||
operation: 'SYNTHESIS',
|
||||
@ -1078,10 +1122,21 @@ export class BomCostingMultimodalAnalyzer {
|
||||
{ contractRetry: contractAttempt > 1 },
|
||||
),
|
||||
maxTokens,
|
||||
timeoutMs,
|
||||
timeoutMs: Math.min(
|
||||
timeoutMs,
|
||||
remainingSynthesisBatchDurationMs,
|
||||
),
|
||||
retrySynthesisTimeout: skuSegment.length === 1,
|
||||
signal: input.signal,
|
||||
})
|
||||
} catch (error) {
|
||||
if (hasReachedSynthesisBatchDeadline()) {
|
||||
throw createSynthesisBatchDeadlineError({
|
||||
batchIndex: batchIndex + 1,
|
||||
totalBatches: skuBatches.length,
|
||||
maxDurationMs: synthesisBatchMaxDurationMs,
|
||||
})
|
||||
}
|
||||
if (
|
||||
skuSegment.length > 1 &&
|
||||
isBomCostingMultimodalTimeoutError(error)
|
||||
|
||||
@ -93,4 +93,33 @@ describe('BOM 多模态网关', () => {
|
||||
'多模态分析服务鉴权失败，请联系管理员检查模型配置',
|
||||
))).toBe(false)
|
||||
})
|
||||
|
||||
it('收到任务总时限取消时保留终止原因，不将其包装成可自动重试的超时', async () => {
|
||||
const controller = new AbortController()
|
||||
const deadlineError = new AppError(
|
||||
ErrorCode.BOM_COSTING_INVALID_PROVIDER,
|
||||
'BOM 核算单次处理已超过 20 分钟，任务已停止',
|
||||
)
|
||||
controller.abort(deadlineError)
|
||||
jest.spyOn(global, 'fetch').mockRejectedValue(
|
||||
new DOMException('The operation was aborted', 'AbortError'),
|
||||
)
|
||||
|
||||
await expect(
|
||||
callBomCostingMultimodalModel({
|
||||
config: {
|
||||
baseUrl: 'https://ai.example.com/v1',
|
||||
apiKey: 'test-key',
|
||||
model: 'gpt-5.6-luna',
|
||||
},
|
||||
operation: 'SYNTHESIS',
|
||||
systemPrompt: 'test',
|
||||
userContent: 'test',
|
||||
maxTokens: 100,
|
||||
timeoutMs: 25,
|
||||
signal: controller.signal,
|
||||
}),
|
||||
).rejects.toBe(deadlineError)
|
||||
expect(isBomCostingMultimodalTransientError(deadlineError)).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
@ -399,6 +399,12 @@ function waitForMultimodalRetry() {
|
||||
})
|
||||
}
|
||||
|
||||
function readTaskDeadlineError(signal: AbortSignal | undefined) {
|
||||
return signal?.aborted && signal.reason instanceof AppError
|
||||
? signal.reason
|
||||
: null
|
||||
}
|
||||
|
||||
export async function callBomCostingMultimodalModel(input: {
|
||||
config: BomCostingMultimodalProviderConfig
|
||||
operation: BomCostingMultimodalOperation
|
||||
@ -407,6 +413,7 @@ export async function callBomCostingMultimodalModel(input: {
|
||||
maxTokens: number
|
||||
timeoutMs: number
|
||||
retrySynthesisTimeout?: boolean
|
||||
signal?: AbortSignal
|
||||
}): Promise<BomCostingMultimodalCallResult> {
|
||||
const endpoint = `${input.config.baseUrl.replace(/\/+$/, '')}/chat/completions`
|
||||
const reasoningEffort = getBomCostingReasoningEffort(
|
||||
@ -422,6 +429,9 @@ export async function callBomCostingMultimodalModel(input: {
|
||||
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
|
||||
let response: Response
|
||||
const timeoutSignal = AbortSignal.timeout(input.timeoutMs)
|
||||
const requestSignal = input.signal
|
||||
? AbortSignal.any([timeoutSignal, input.signal])
|
||||
: timeoutSignal
|
||||
try {
|
||||
response = await fetch(endpoint, {
|
||||
method: 'POST',
|
||||
@ -442,9 +452,11 @@ export async function callBomCostingMultimodalModel(input: {
|
||||
...(reasoningEffort ? { reasoning_effort: reasoningEffort } : {}),
|
||||
}),
|
||||
cache: 'no-store',
|
||||
signal: timeoutSignal,
|
||||
signal: requestSignal,
|
||||
})
|
||||
} catch (error) {
|
||||
const taskDeadlineError = readTaskDeadlineError(requestSignal)
|
||||
if (taskDeadlineError) throw taskDeadlineError
|
||||
console.error('[BOM costing] multimodal request failed', {
|
||||
operation: input.operation,
|
||||
model: input.config.model,
|
||||
@ -510,9 +522,11 @@ export async function callBomCostingMultimodalModel(input: {
|
||||
let result: Awaited<ReturnType<typeof readStreamingChatCompletion>>
|
||||
try {
|
||||
result = await readStreamingChatCompletion(response, {
|
||||
signal: timeoutSignal,
|
||||
signal: requestSignal,
|
||||
})
|
||||
} catch (error) {
|
||||
const taskDeadlineError = readTaskDeadlineError(requestSignal)
|
||||
if (taskDeadlineError) throw taskDeadlineError
|
||||
console.error('[BOM costing] multimodal response stream failed', {
|
||||
operation: input.operation,
|
||||
model: input.config.model,
|
||||
|
||||
@ -304,7 +304,7 @@ async function persistArchiveProgress(
|
||||
}
|
||||
|
||||
export class BomCostingProcessingService {
|
||||
async processCase(caseId: string) {
|
||||
async processCase(caseId: string, options: { signal?: AbortSignal } = {}) {
|
||||
const claimed = await bomCostingRepository.claimCaseForProcessing(caseId)
|
||||
if (!claimed) return
|
||||
|
||||
@ -325,6 +325,9 @@ export class BomCostingProcessingService {
|
||||
let factAnalysisOwnedByTask = false
|
||||
let factAnalysisId: string | null = null
|
||||
try {
|
||||
if (options.signal?.aborted) {
|
||||
throw options.signal.reason ?? new Error('BOM 核算任务已取消')
|
||||
}
|
||||
const record = await bomCostingRepository.findCaseForProcessing(caseId)
|
||||
if (!record) throw new Error('BOM 核算案例不存在')
|
||||
sourceSnapshotId = record.sourceSnapshotId
|
||||
@ -785,6 +788,7 @@ export class BomCostingProcessingService {
|
||||
analysisRun.resultSnapshot,
|
||||
),
|
||||
analysisInstruction: analysisRun.instruction,
|
||||
signal: options.signal,
|
||||
onFactBatchStarted:
|
||||
factAnalysisOwnedByTask && factAnalysisId
|
||||
? async (progress) => {
|
||||
|
||||
@ -39,6 +39,7 @@ export interface CompetitorAnalysisProvider {
|
||||
existingFactSnapshot?: BomCostingSourceFactSnapshot | null
|
||||
synthesisCheckpoint?: BomCostingSynthesisCheckpoint | null
|
||||
analysisInstruction?: string
|
||||
signal?: AbortSignal
|
||||
onFactBatchStarted?: (
|
||||
progress: Omit<BomCostingFactBatchProgress, 'snapshot' | 'call'>,
|
||||
) => Promise<void>
|
||||
@ -329,6 +330,7 @@ class MarketDetailMultimodalProvider implements CompetitorAnalysisProvider {
|
||||
existingFactSnapshot?: BomCostingSourceFactSnapshot | null
|
||||
synthesisCheckpoint?: BomCostingSynthesisCheckpoint | null
|
||||
analysisInstruction?: string
|
||||
signal?: AbortSignal
|
||||
onFactBatchStarted?: (
|
||||
progress: Omit<BomCostingFactBatchProgress, 'snapshot' | 'call'>,
|
||||
) => Promise<void>
|
||||
|
||||
@ -56,4 +56,16 @@ describe('BomCostingWorkerService', () => {
|
||||
expect(bomCostingRepository.findNextQueuedCaseForWorker).not.toHaveBeenCalled()
|
||||
expect(mockProcessCase).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('将单轮任务总时限传给实际处理链路', async () => {
|
||||
jest.spyOn(bomCostingRepository, 'findNextQueuedCaseForWorker')
|
||||
.mockResolvedValue({ id: 'case-deadline' } as never)
|
||||
const controller = new AbortController()
|
||||
|
||||
await new BomCostingWorkerService().runNext({ signal: controller.signal })
|
||||
|
||||
expect(mockProcessCase).toHaveBeenCalledWith('case-deadline', {
|
||||
signal: controller.signal,
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@ -6,7 +6,7 @@ import { bomCostingProcessingService } from './bom-costing-processing.service'
|
||||
export const BOM_COSTING_STALE_PROCESSING_MS = 10 * 60_000
|
||||
|
||||
export class BomCostingWorkerService {
|
||||
async runNext() {
|
||||
async runNext(input: { signal?: AbortSignal } = {}) {
|
||||
const staleBefore = new Date(Date.now() - BOM_COSTING_STALE_PROCESSING_MS)
|
||||
const [recoveredSourceSnapshots, recoveredCaseIds] = await Promise.all([
|
||||
bomCostingRepository.recoverStaleSourceSnapshots({ staleBefore }),
|
||||
@ -34,7 +34,13 @@ export class BomCostingWorkerService {
|
||||
}
|
||||
}
|
||||
|
||||
await bomCostingProcessingService.processCase(nextCase.id)
|
||||
if (input.signal) {
|
||||
await bomCostingProcessingService.processCase(nextCase.id, {
|
||||
signal: input.signal,
|
||||
})
|
||||
} else {
|
||||
await bomCostingProcessingService.processCase(nextCase.id)
|
||||
}
|
||||
return {
|
||||
status: 'processed' as const,
|
||||
caseId: nextCase.id,
|
||||
|
||||
@ -1,7 +1,11 @@
|
||||
import * as mysql from 'mysql2/promise'
|
||||
import type { ResultSetHeader, RowDataPacket } from 'mysql2/promise'
|
||||
import type { AuthUser } from '@/server/core/types'
|
||||
import { pickCurrentOverdueStageMetric } from '@/lib/oa-board-overtime'
|
||||
import {
|
||||
listHistoricalOverdueStageMetrics,
|
||||
pickCurrentOverdueStageMetric,
|
||||
type OaBoardTimelineMetricLike,
|
||||
} from '@/lib/oa-board-overtime'
|
||||
import { createPlanningConnectionConfig } from './oa-board-sync.service'
|
||||
|
||||
const OA_BOARD_TIMEOUT_FOLLOWUP_TABLE = 'oa_board_timeout_followup'
|
||||
@ -139,12 +143,29 @@ type TimeoutFollowupPermissionSnapshotRow = RowDataPacket & {
|
||||
export type OaBoardTimeoutFollowupEditPermission = {
|
||||
allowed: boolean
|
||||
message?: string
|
||||
mode?: 'CURRENT' | 'HISTORICAL'
|
||||
stageLabel?: string
|
||||
targetDays?: number
|
||||
actualDays?: number
|
||||
overtimeDays?: number
|
||||
}
|
||||
|
||||
type OaBoardTimeoutFollowupAssessment = {
|
||||
mode: 'CURRENT' | 'HISTORICAL'
|
||||
stageLabel: string
|
||||
targetDays: number
|
||||
actualDays: number
|
||||
overtimeDays: number
|
||||
ownerLabel: string | null
|
||||
}
|
||||
|
||||
type SavedTimeoutFollowupAssessment = RowDataPacket & {
|
||||
stageLabel: string
|
||||
targetDays: string | number
|
||||
actualDays: string | number
|
||||
overtimeDays: string | number
|
||||
}
|
||||
|
||||
function asNullableText(value: string | null | undefined) {
|
||||
const text = String(value ?? '').trim()
|
||||
return text || null
|
||||
@ -187,6 +208,17 @@ function readJsonRecord(value: string | null) {
|
||||
}
|
||||
}
|
||||
|
||||
function readJsonList(value: string | null) {
|
||||
if (!value) return []
|
||||
|
||||
try {
|
||||
const parsed: unknown = JSON.parse(value)
|
||||
return Array.isArray(parsed) ? parsed : []
|
||||
} catch {
|
||||
return []
|
||||
}
|
||||
}
|
||||
|
||||
function readFormSummary(snapshot: Record<string, unknown> | null) {
|
||||
const metadata = snapshot?.metadata
|
||||
if (!metadata || typeof metadata !== 'object' || Array.isArray(metadata)) return {}
|
||||
@ -209,7 +241,37 @@ function buildCurrentNodeOvertimeStageKey(input: {
|
||||
return `${String(input.flowType ?? '').trim()}:${String(input.nodeName ?? '').trim()}`
|
||||
}
|
||||
|
||||
function readCurrentOverdueAssessment(snapshot: Record<string, unknown> | null, stageKey: string) {
|
||||
function readTimelineMetrics(
|
||||
snapshot: Record<string, unknown> | null,
|
||||
serializedTimelineMetrics?: string | null,
|
||||
): OaBoardTimelineMetricLike[] {
|
||||
const metadata = snapshot?.metadata
|
||||
const rawTimelineMetrics =
|
||||
metadata && typeof metadata === 'object' && !Array.isArray(metadata)
|
||||
? (metadata as Record<string, unknown>).timelineMetrics
|
||||
: null
|
||||
const source = Array.isArray(rawTimelineMetrics)
|
||||
? rawTimelineMetrics
|
||||
: readJsonList(serializedTimelineMetrics ?? null)
|
||||
|
||||
return source.flatMap((metric) => {
|
||||
if (!metric || typeof metric !== 'object' || Array.isArray(metric)) return []
|
||||
const record = metric as Record<string, unknown>
|
||||
const key = readText(record, 'key')
|
||||
const label = readText(record, 'label')
|
||||
const targetDays = Number(record.targetDays)
|
||||
const actualDays = Number(record.actualDays)
|
||||
if (!key || !label || !Number.isFinite(targetDays) || !Number.isFinite(actualDays)) return []
|
||||
|
||||
return [{ key, label, targetDays, actualDays }]
|
||||
})
|
||||
}
|
||||
|
||||
function readCurrentOverdueAssessment(
|
||||
snapshot: Record<string, unknown> | null,
|
||||
timelineMetrics: readonly OaBoardTimelineMetricLike[],
|
||||
stageKey: string,
|
||||
): OaBoardTimeoutFollowupAssessment | null {
|
||||
const formSummary = readFormSummary(snapshot)
|
||||
const stageNames = splitStageValues(readText(formSummary, 'approvalStageDisplay'))
|
||||
const currentNodeNames = splitStageValues(readText(formSummary, 'currentNodeDisplay'))
|
||||
@ -264,6 +326,7 @@ function readCurrentOverdueAssessment(snapshot: Record<string, unknown> | null,
|
||||
const taskRecord = taskRecords[0]
|
||||
if (taskRecord) {
|
||||
return {
|
||||
mode: 'CURRENT',
|
||||
stageLabel: `${taskRecord.stageName || taskRecord.processName || '-'} · ${taskRecord.nodeName || stageKey}`,
|
||||
targetDays: Number(taskRecord.targetHours.toFixed(2)),
|
||||
actualDays: Number(taskRecord.actualHours.toFixed(2)),
|
||||
@ -272,27 +335,11 @@ function readCurrentOverdueAssessment(snapshot: Record<string, unknown> | null,
|
||||
}
|
||||
}
|
||||
|
||||
const metadata = snapshot?.metadata
|
||||
const rawTimelineMetrics =
|
||||
metadata && typeof metadata === 'object' && !Array.isArray(metadata)
|
||||
? (metadata as Record<string, unknown>).timelineMetrics
|
||||
: []
|
||||
const timelineMetrics = Array.isArray(rawTimelineMetrics)
|
||||
? rawTimelineMetrics.flatMap((metric) => {
|
||||
if (!metric || typeof metric !== 'object' || Array.isArray(metric)) return []
|
||||
const record = metric as Record<string, unknown>
|
||||
const key = readText(record, 'key')
|
||||
const label = readText(record, 'label')
|
||||
const targetDays = Number(record.targetDays)
|
||||
const actualDays = Number(record.actualDays)
|
||||
if (!key || !label || !Number.isFinite(targetDays) || !Number.isFinite(actualDays)) return []
|
||||
return [{ key, label, targetDays, actualDays }]
|
||||
})
|
||||
: []
|
||||
const overdueStage = pickCurrentOverdueStageMetric(stageNames, timelineMetrics)
|
||||
if (!overdueStage || overdueStage.metricKey !== stageKey) return null
|
||||
|
||||
return {
|
||||
mode: 'CURRENT',
|
||||
stageLabel: `${overdueStage.stageName} · ${currentNodeNames[0] || '-'}`,
|
||||
targetDays: overdueStage.targetDays,
|
||||
actualDays: overdueStage.actualDays,
|
||||
@ -301,6 +348,48 @@ function readCurrentOverdueAssessment(snapshot: Record<string, unknown> | null,
|
||||
}
|
||||
}
|
||||
|
||||
function readHistoricalOverdueAssessment(
|
||||
snapshot: Record<string, unknown> | null,
|
||||
timelineMetrics: readonly OaBoardTimelineMetricLike[],
|
||||
stageKey: string,
|
||||
): OaBoardTimeoutFollowupAssessment | null {
|
||||
const formSummary = readFormSummary(snapshot)
|
||||
const stageNames = splitStageValues(readText(formSummary, 'approvalStageDisplay'))
|
||||
const historicalStage = listHistoricalOverdueStageMetrics(stageNames, timelineMetrics).find(
|
||||
(stage) => stage.metricKey === stageKey,
|
||||
)
|
||||
if (!historicalStage) return null
|
||||
|
||||
return {
|
||||
mode: 'HISTORICAL',
|
||||
stageLabel: historicalStage.metricLabel,
|
||||
targetDays: historicalStage.targetDays,
|
||||
actualDays: historicalStage.actualDays,
|
||||
overtimeDays: historicalStage.overtimeDays,
|
||||
ownerLabel: null,
|
||||
}
|
||||
}
|
||||
|
||||
function readSavedHistoricalAssessment(
|
||||
row: SavedTimeoutFollowupAssessment | undefined,
|
||||
): OaBoardTimeoutFollowupAssessment | null {
|
||||
if (!row) return null
|
||||
|
||||
const targetDays = toNumber(row.targetDays)
|
||||
const actualDays = toNumber(row.actualDays)
|
||||
const overtimeDays = toNumber(row.overtimeDays)
|
||||
if (!row.stageLabel.trim() || overtimeDays <= 0) return null
|
||||
|
||||
return {
|
||||
mode: 'HISTORICAL',
|
||||
stageLabel: row.stageLabel.trim(),
|
||||
targetDays,
|
||||
actualDays,
|
||||
overtimeDays,
|
||||
ownerLabel: null,
|
||||
}
|
||||
}
|
||||
|
||||
function isFullAccessEditor(user: AuthUser | undefined) {
|
||||
if (!user) return false
|
||||
if (user.role === 'SUPER_ADMIN' || user.role === 'ADMIN') return true
|
||||
@ -424,12 +513,37 @@ export class OaBoardTimeoutFollowupService {
|
||||
|
||||
const snapshot = readJsonRecord(row.raw_snapshot)
|
||||
const formSummary = readFormSummary(snapshot)
|
||||
const currentOverdueStage = readCurrentOverdueAssessment(snapshot, input.stageKey)
|
||||
if (!currentOverdueStage) {
|
||||
return { allowed: false, message: '当前审批节点未超时或已变化，无法保存超时跟进' }
|
||||
const timelineMetrics = readTimelineMetrics(snapshot, row.timeline_metrics)
|
||||
const currentOverdueStage = readCurrentOverdueAssessment(
|
||||
snapshot,
|
||||
timelineMetrics,
|
||||
input.stageKey,
|
||||
)
|
||||
const historicalOverdueStage = currentOverdueStage
|
||||
? null
|
||||
: readHistoricalOverdueAssessment(snapshot, timelineMetrics, input.stageKey)
|
||||
|
||||
await ensureTable(connection)
|
||||
const [savedRows] = await connection.query<SavedTimeoutFollowupAssessment[]>(
|
||||
`
|
||||
SELECT stage_label AS stageLabel, target_days AS targetDays, actual_days AS actualDays,
|
||||
overtime_days AS overtimeDays
|
||||
FROM ${OA_BOARD_TIMEOUT_FOLLOWUP_TABLE}
|
||||
WHERE board_key = ? AND stage_key = ?
|
||||
LIMIT 1
|
||||
`,
|
||||
[input.boardKey.trim(), input.stageKey.trim()],
|
||||
)
|
||||
const savedHistoricalStage =
|
||||
currentOverdueStage || historicalOverdueStage
|
||||
? null
|
||||
: readSavedHistoricalAssessment(savedRows[0])
|
||||
const overdueStage = currentOverdueStage ?? historicalOverdueStage ?? savedHistoricalStage
|
||||
if (!overdueStage) {
|
||||
return { allowed: false, message: '该节点未超时，或流程数据已变化，无法保存超时跟进' }
|
||||
}
|
||||
|
||||
if (isFullAccessEditor(user)) return { allowed: true, ...currentOverdueStage }
|
||||
if (isFullAccessEditor(user)) return { allowed: true, ...overdueStage }
|
||||
|
||||
const fieldValues: Record<string, string | null> = {
|
||||
productPlanner: readText(formSummary, 'productPlanner') || row.product_planner,
|
||||
@ -442,12 +556,21 @@ export class OaBoardTimeoutFollowupService {
|
||||
}
|
||||
const actorName = user.name.trim()
|
||||
const productPlanners = splitUserNames(fieldValues.productPlanner)
|
||||
if (productPlanners.includes(actorName)) return { allowed: true, ...overdueStage }
|
||||
|
||||
if (overdueStage.mode === 'HISTORICAL') {
|
||||
return {
|
||||
allowed: false,
|
||||
message: '历史超时仅管理员、荆川或商品企划可保存跟进',
|
||||
}
|
||||
}
|
||||
|
||||
const currentOwners = splitUserNames(readText(formSummary, 'currentOwnerDisplay'))
|
||||
const stageOwners = currentOverdueStage.ownerLabel ? splitUserNames(currentOverdueStage.ownerLabel) : []
|
||||
const stageOwners = overdueStage.ownerLabel ? splitUserNames(overdueStage.ownerLabel) : []
|
||||
const mergedOwners = Array.from(new Set([...currentOwners, ...stageOwners]))
|
||||
|
||||
if (productPlanners.includes(actorName) || mergedOwners.includes(actorName)) {
|
||||
return { allowed: true, ...currentOverdueStage }
|
||||
if (mergedOwners.includes(actorName)) {
|
||||
return { allowed: true, ...overdueStage }
|
||||
}
|
||||
|
||||
return { allowed: false, message: '仅管理员、荆川、商品企划或当前节点负责人可保存跟进' }
|
||||
|
||||
@ -212,13 +212,15 @@ start_pm2_release() {
|
||||
prepare_standalone_runtime "$app_root"
|
||||
# Read the shared environment before resolving the curl limit. The worker
|
||||
# command itself sources this file later, but doing it only there leaves the
|
||||
# already-interpolated --max-time stuck at the 45-minute default.
|
||||
# already-interpolated --max-time stuck at an unsafe long default.
|
||||
set -a
|
||||
source "$SHARED/.env"
|
||||
set +a
|
||||
worker_timeout="${BOM_COSTING_WORKER_REQUEST_TIMEOUT_SECONDS:-2700}"
|
||||
if ! [[ "$worker_timeout" =~ ^[1-9][0-9]*$ ]]; then
|
||||
worker_timeout=2700
|
||||
worker_timeout="${BOM_COSTING_WORKER_REQUEST_TIMEOUT_SECONDS:-1200}"
|
||||
# One worker turn contains its own per-request limits. Never let a stale
|
||||
# environment value turn a failed model call into hours of silent waiting.
|
||||
if ! [[ "$worker_timeout" =~ ^[1-9][0-9]*$ ]] || [ "$worker_timeout" -lt 60 ] || [ "$worker_timeout" -gt 1200 ]; then
|
||||
worker_timeout=1200
|
||||
fi
|
||||
start_command="cd '$app_root' && set -a && source '$SHARED/.env' && set +a && exec env PORT=$APP_PORT HOSTNAME=0.0.0.0 node .next/standalone/server.js"
|
||||
worker_command="cd '$app_root' && set -a && source '$SHARED/.env' && set +a && : \"\${BOM_COSTING_WORKER_SECRET:?BOM_COSTING_WORKER_SECRET is required}\"; while true; do curl --fail --silent --show-error --max-time $worker_timeout --request POST 'http://127.0.0.1:$APP_PORT/api/internal/bom-costing/worker/run' --header \"x-bom-costing-worker-secret: \$BOM_COSTING_WORKER_SECRET\" >/dev/null || echo '[BOM costing worker] loop request failed' >&2; sleep 10; done"
|
||||
|
||||
@ -289,14 +289,16 @@ start_pm2_release() {
|
||||
prepare_standalone_runtime "$app_root"
|
||||
# Read the shared environment before resolving the curl limit. The worker
|
||||
# command itself sources this file later, but doing it only there leaves the
|
||||
# already-interpolated --max-time stuck at the 45-minute default.
|
||||
# already-interpolated --max-time stuck at an unsafe long default.
|
||||
set -a
|
||||
source "$SHARED/.env"
|
||||
set +a
|
||||
ensure_image_search_model
|
||||
worker_timeout="${BOM_COSTING_WORKER_REQUEST_TIMEOUT_SECONDS:-2700}"
|
||||
if ! [[ "$worker_timeout" =~ ^[1-9][0-9]*$ ]]; then
|
||||
worker_timeout=2700
|
||||
worker_timeout="${BOM_COSTING_WORKER_REQUEST_TIMEOUT_SECONDS:-1200}"
|
||||
# One worker turn contains its own per-request limits. Never let a stale
|
||||
# environment value turn a failed model call into hours of silent waiting.
|
||||
if ! [[ "$worker_timeout" =~ ^[1-9][0-9]*$ ]] || [ "$worker_timeout" -lt 60 ] || [ "$worker_timeout" -gt 1200 ]; then
|
||||
worker_timeout=1200
|
||||
fi
|
||||
start_command="cd '$app_root' && set -a && source '$SHARED/.env' && set +a && exec env IMAGE_SEARCH_MODEL_PATH='$SHARED/models/dinov2-small' PORT=$APP_PORT HOSTNAME=0.0.0.0 node .next/standalone/server.js"
|
||||
worker_command="cd '$app_root' && set -a && source '$SHARED/.env' && set +a && : \"\${BOM_COSTING_WORKER_SECRET:?BOM_COSTING_WORKER_SECRET is required}\"; while true; do curl --fail --silent --show-error --max-time $worker_timeout --request POST 'http://127.0.0.1:$APP_PORT/api/internal/bom-costing/worker/run' --header \"x-bom-costing-worker-secret: \$BOM_COSTING_WORKER_SECRET\" >/dev/null || echo '[BOM costing worker] loop request failed' >&2; sleep 10; done"
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user