From 52b55df7b27969d5789a719029e67f8fff7caea5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Mon, 14 Sep 2026 17:56:20 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E4=BB=BB=E5=8A=A1=E5=88=A4=E6=AD=BB):=20?= =?UTF-8?q?=E5=BF=83=E8=B7=B3=E6=AD=A3=E5=B8=B8=E4=BD=86=20180=20=E5=88=86?= =?UTF-8?q?=E9=92=9F=E6=97=A0=E7=BB=93=E6=9E=9C=E4=B8=8A=E6=8A=A5=E7=9A=84?= =?UTF-8?q?=E4=BA=8C=E6=AC=A1=E5=88=A4=E6=AD=BB=E7=BA=BF=EF=BC=8813=20?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=EF=BC=89+=20=E5=90=8C=E6=9C=9F=E5=BE=85?= =?UTF-8?q?=E5=8F=91=E6=94=B9=E5=8A=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 判死线(治 28131 型「主线程卡死、心跳线程照发」): - 判据改看 biz_task_scope_state.last_chunk_at(HTTP 心跳不刷新它);从未上报跳过不判 - 中央线覆盖 DELETE_BRAND/PRODUCT_RISK_RESOLVE/PRICE_TRACK/SHOP_MATCH/PATROL_DELETE/QUERY_ASIN/WITHDRAW - 自带线接入 COLLECT_DATA/SIMILAR_ASIN/APPEARANCE_PATENT/SHOP_DATA_CRAWL/PUBLISH/BRAND - 客户端心跳带处理位置 progressText,判死文案含最后位置;no-result-upload-timeout-minutes 默认 180(0 关闭) 同期带上另一工作流的待发改动:跟价换 IP 重试、品牌检测重试上限与 LLM 并发下调、 教程包后台管理页与 V126 迁移、admin-vue 教程记录页。 --- .../e2e/records-tutorial.spec.ts | 79 ++++++++ .../scripts/mock-admin-server.mjs | 14 ++ .../src/pages/records/RecordsTutorialPage.vue | 127 +++++++++++- .../src/pages/records/tutorial-api.ts | 15 +- .../src/pages/records/tutorial-dto.ts | 2 + .../src/pages/records/tutorial-model.ts | 1 + .../tests/align-tutorial.test.ts | 15 +- .../config/AppearancePatentProperties.java | 13 +- .../aiimage/config/BrandCheckProperties.java | 15 +- .../config/BrandProgressProperties.java | 7 + .../config/DeleteBrandProgressProperties.java | 13 ++ .../aiimage/config/SimilarAsinProperties.java | 8 + .../service/AppearancePatentTaskService.java | 61 ++++++ .../brand/client/BrandCheckClient.java | 28 ++- .../BrandTaskProgressCacheService.java | 2 + .../brand/service/BrandTaskService.java | 49 +++++ .../service/CollectDataService.java | 86 ++++++++ .../service/DeleteBrandStaleTaskService.java | 138 ++++++++++++- .../publish/service/PublishTaskService.java | 93 ++++++++- .../service/ShopDataCrawlTaskService.java | 86 +++++++- .../service/SimilarAsinTaskService.java | 54 +++++ .../controller/TaskHeartbeatController.java | 4 + .../task/mapper/TaskScopeStateMapper.java | 25 +++ .../task/model/dto/TaskHeartbeatRequest.java | 4 + .../task/model/dto/TaskScopeLastChunkDto.java | 12 ++ .../service/TaskHeartbeatPositionService.java | 98 +++++++++ .../controller/TutorialAdminController.java | 17 +- .../model/entity/TutorialPackageEntity.java | 3 + .../service/TutorialPackageService.java | 42 +++- .../resources/application-local.example.yml | 2 +- .../src/main/resources/application.yml | 14 +- .../db/V126__tutorial_package_version.sql | 22 +++ .../service/CollectDataNoUploadStaleTest.java | 143 ++++++++++++++ .../DeleteBrandStaleTaskServiceTest.java | 186 +++++++++++++++++- .../service/TaskModuleCoverageTest.java | 3 +- .../TaskHeartbeatPositionServiceTest.java | 114 +++++++++++ 36 files changed, 1538 insertions(+), 57 deletions(-) create mode 100644 admin-frontend-vue/e2e/records-tutorial.spec.ts create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskScopeLastChunkDto.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatPositionService.java create mode 100644 backend-java/src/main/resources/db/V126__tutorial_package_version.sql create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataNoUploadStaleTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatPositionServiceTest.java diff --git a/admin-frontend-vue/e2e/records-tutorial.spec.ts b/admin-frontend-vue/e2e/records-tutorial.spec.ts new file mode 100644 index 00000000..19708386 --- /dev/null +++ b/admin-frontend-vue/e2e/records-tutorial.spec.ts @@ -0,0 +1,79 @@ +import { expect, test, type Page } from '@playwright/test' + +// 教程管理页验收(module 记录与版本):列表新增「版本号」列 + 上传时间列, +// 默认按上传时间降序;版本号/上传时间表头可点击切换升降序;空版本历史行沉底。 +// 依赖 scripts/mock-admin-server.mjs 的教程包夹具(4 条,其中 1 条无版本号、1 条 09-01 历史行)。 + +async function openTutorial(page: Page) { + await page.goto('/admin-vue/records/tutorial') + await expect(page.locator('.admin-topbar h1')).toHaveText('教程管理') + await expect(page.locator('.panel-box tbody tr').first()).toBeVisible() +} + +const versionTexts = (page: Page) => + page.locator('.panel-box tbody tr td:nth-child(3)').allTextContents() + +const fileTexts = (page: Page) => + page.locator('.panel-box tbody tr td:nth-child(2) .file-name').allTextContents() + +test('test_tutorial_list_default_desc_by_upload_time', async ({ page }) => { + await openTutorial(page) + + // 表头:版本号、上传时间(均带排序标记) + await expect(page.locator('.sort-version')).toHaveText(/版本号/) + await expect(page.locator('.sort-time')).toHaveText(/上传时间/) + // 默认排序状态:上传时间降序(▼),版本号未激活(▲▼) + await expect(page.locator('.sort-time .sort-mark')).toHaveText('▼') + await expect(page.locator('.sort-version .sort-mark')).toHaveText('▲▼') + + // 默认按上传时间降序:09-14 → 09-12 → 09-10 → 09-01;空版本行沉底 + expect(await versionTexts(page)).toEqual(['v3.10.0', 'v3.2.0', 'v3.1.0', '—']) + expect(await fileTexts(page)).toEqual([ + '数富AI-教学客户端-v4.zip', + '数富AI-教学客户端-v3.zip', + '数富AI-教学客户端-v2.zip', + '数富AI-教学客户端.zip', + ]) + // 当前生效 = 最新上传(夹具里 09-14 那条),不随列表排序变化 + await expect(page.locator('.tag-active')).toHaveCount(1) + await expect(page.locator('.tag-active').locator('xpath=..')).toHaveText(/数富AI-教学客户端-v4\.zip/) +}) + +test('test_tutorial_sort_toggle_by_version_and_time', async ({ page }) => { + await openTutorial(page) + + // 点「版本号」:首次为降序(v3.10.0 按自然序大于 v3.2.0),空版本行仍在末尾 + await page.locator('.sort-version').click() + await expect(page.locator('.sort-version .sort-mark')).toHaveText('▼') + expect(await versionTexts(page)).toEqual(['v3.10.0', 'v3.2.0', 'v3.1.0', '—']) + + // 再点一次切升序 + await page.locator('.sort-version').click() + await expect(page.locator('.sort-version .sort-mark')).toHaveText('▲') + expect(await versionTexts(page)).toEqual(['v3.1.0', 'v3.2.0', 'v3.10.0', '—']) + + // 点回「上传时间」:切换字段时回到降序默认 + await page.locator('.sort-time').click() + await expect(page.locator('.sort-time .sort-mark')).toHaveText('▼') + await expect(page.locator('.sort-version .sort-mark')).toHaveText('▲▼') + expect(await fileTexts(page)).toEqual([ + '数富AI-教学客户端-v4.zip', + '数富AI-教学客户端-v3.zip', + '数富AI-教学客户端-v2.zip', + '数富AI-教学客户端.zip', + ]) +}) + +test('test_tutorial_upload_dialog_has_version_field', async ({ page }) => { + await openTutorial(page) + + await page.getByRole('button', { name: '上传教程包' }).click() + const dialog = page.locator('.el-dialog') + await expect(dialog).toBeVisible() + await expect(dialog.locator('.el-form-item').first()).toContainText('版本号') + await expect(dialog.locator('input').first()).toHaveAttribute('maxlength', '64') + + // 不选文件直接提交:提示仍以 zip 为必填(版本号可空) + await dialog.getByRole('button', { name: '上传教程包' }).click() + await expect(dialog.locator('.el-alert')).toContainText('请选择 zip 压缩包') +}) diff --git a/admin-frontend-vue/scripts/mock-admin-server.mjs b/admin-frontend-vue/scripts/mock-admin-server.mjs index 4c4db56f..51424e8a 100644 --- a/admin-frontend-vue/scripts/mock-admin-server.mjs +++ b/admin-frontend-vue/scripts/mock-admin-server.mjs @@ -142,6 +142,15 @@ function json(res, payload, status = 200) { res.end(body) } +/* ===================== 教程包夹具(教程管理页:版本号列 + 上传时间排序验收用) ===================== */ +// 故意乱序给出,且含一条无版本号的历史行:页面默认应按上传时间降序、空版本行沉底。 +const TUTORIAL_PACKAGES = [ + { id: 3, file_name: '数富AI-教学客户端-v3.zip', version: 'v3.2.0', object_key: 'tutorial/20260912090000-数富AI-教学客户端-v3.zip', file_size: 1048576, file_url: 'https://oss.aishufu.top/client/tutorial/t3.zip', created_at: '2026-09-12 09:00' }, + { id: 1, file_name: '数富AI-教学客户端.zip', version: '', object_key: 'tutorial/数富AI-教学客户端.zip', file_size: 0, file_url: 'https://oss.aishufu.top/client/tutorial/legacy.zip', created_at: '2026-09-01 08:00' }, + { id: 4, file_name: '数富AI-教学客户端-v4.zip', version: 'v3.10.0', object_key: 'tutorial/20260914103000-数富AI-教学客户端-v4.zip', file_size: 2097152, file_url: 'https://oss.aishufu.top/client/tutorial/t4.zip', created_at: '2026-09-14 10:30' }, + { id: 2, file_name: '数富AI-教学客户端-v2.zip', version: 'v3.1.0', object_key: 'tutorial/20260910120000-数富AI-教学客户端-v2.zip', file_size: 524288, file_url: 'https://oss.aishufu.top/client/tutorial/t2.zip', created_at: '2026-09-10 12:00' }, +] + /* ===================== 站内通知夹具(铃铛面板:搜索/按天分组/分页验收用) ===================== */ const NOTIFICATIONS = [ ...[6, 5, 4, 3, 2, 1].map((seq) => notif(seq, '2026-09-13', 22, seq, seq <= 3)), @@ -243,6 +252,11 @@ const server = createServer((req, res) => { if (url === '/api/admin/notifications') { return json(res, notificationPage(new URL(req.url || '/', 'http://127.0.0.1').searchParams)) } + if (url === '/api/admin/tutorials') { + // 与 Java 侧同契约:created_at 降序返回(页面"当前生效"取第一条)。 + const items = [...TUTORIAL_PACKAGES].sort((a, b) => (a.created_at < b.created_at ? 1 : -1)) + return json(res, { success: true, data: { items } }) + } if (url.startsWith('/api/')) { return json(res, { success: true, data: { items: [], total: 0 } }) } diff --git a/admin-frontend-vue/src/pages/records/RecordsTutorialPage.vue b/admin-frontend-vue/src/pages/records/RecordsTutorialPage.vue index 75f52c5d..c9e279c9 100644 --- a/admin-frontend-vue/src/pages/records/RecordsTutorialPage.vue +++ b/admin-frontend-vue/src/pages/records/RecordsTutorialPage.vue @@ -2,7 +2,8 @@ import { formatDateTime } from '@/utils/datetime' /** 教程管理页:工具台首页「立即下载教程」的包体管理(上传新包 / 列表 / 下载 / 删除)。 * 上传走浏览器直传 MinIO(presign → PUT 带进度 → confirm 落库),与软件版本管理页同一套链路; - * 工具台始终下载"最新上传"的包(列表第一条即当前生效)。 */ + * 上传时可填版本号(仅展示与排序用);工具台始终下载"最新上传"的包(不随列表排序变化)。 + * 列表默认按上传时间降序,「版本号」「上传时间」表头可点击切换升降序。 */ import { computed, onMounted, ref, watch } from 'vue' import OldPagination from '@/components/OldPagination.vue' import { ElMessage } from 'element-plus' @@ -22,12 +23,64 @@ const filteredItems = computed(() => { /** 全站分页统一:客户端分页(10/20/50/100)。 */ const page = ref(1) const pageSize = ref(10) -const pagedItems = computed(() => filteredItems.value.slice((page.value - 1) * pageSize.value, page.value * pageSize.value)) + +/** 排序:默认按上传时间降序(最新上传在最前,与后端返回顺序一致);点表头切换升降序。 */ +type SortKey = 'version' | 'createdAt' +const sortKey = ref('createdAt') +const sortAsc = ref(false) + +function toggleSort(key: SortKey) { + if (sortKey.value === key) { + sortAsc.value = !sortAsc.value + } else { + sortKey.value = key + sortAsc.value = false + } + page.value = 1 +} + +/** 排序标记:未激活 ▲▼,激活时只留方向箭头(比 ⇅/↕ 字形支持好,避免 Windows 缺字形显示方框)。 */ +function sortMark(key: SortKey): string { + if (sortKey.value !== key) return '▲▼' + return sortAsc.value ? '▲' : '▼' +} + +function compareText(a: string, b: string): number { + return a.localeCompare(b, 'zh-Hans-CN', { numeric: true, sensitivity: 'base' }) +} + +const sortedItems = computed(() => { + const rows = [...filteredItems.value] + rows.sort((a, b) => { + // 版本号为空的历史行始终排在末尾,避免切换排序时"无版本"占满首页。 + if (sortKey.value === 'version') { + const av = a.version + const bv = b.version + if (!av || !bv) { + if (!av && !bv) return 0 + return av ? -1 : 1 + } + const diff = compareText(av, bv) + return sortAsc.value ? diff : -diff + } + const at = a.createdAt + const bt = b.createdAt + if (!at || !bt) { + if (!at && !bt) return 0 + return at ? -1 : 1 + } + const diff = compareText(at, bt) + return sortAsc.value ? diff : -diff + }) + return rows +}) + +const pagedItems = computed(() => sortedItems.value.slice((page.value - 1) * pageSize.value, page.value * pageSize.value)) function changePage(p: number) { page.value = p } function changeSize(size: number) { pageSize.value = size; page.value = 1 } -// 搜索导致数据收缩时回钳页码,避免停在空页。 -watch(filteredItems, () => { - page.value = Math.min(page.value, Math.max(1, Math.ceil(filteredItems.value.length / pageSize.value))) +// 搜索/排序导致数据收缩时回钳页码,避免停在空页。 +watch(sortedItems, () => { + page.value = Math.min(page.value, Math.max(1, Math.ceil(sortedItems.value.length / pageSize.value))) }) /** 当前生效包 = 列表第一条(工具台下发的就是它)。 */ @@ -80,6 +133,7 @@ async function removeOne(row: TutorialPackageItem) { const uploadVisible = ref(false) const uploading = ref(false) const uploadPercent = ref(0) +const newVersion = ref('') const pickedFile = ref(null) /** 弹窗内成功/失败文案留驻(对齐软件版本页交互)。 */ const uploadMsg = ref('') @@ -103,6 +157,7 @@ function onFileChange(file: File) { function openUpload() { uploadMsg.value = '' uploadMsgOk.value = false + newVersion.value = '' pickedFile.value = null uploadVisible.value = true } @@ -126,12 +181,13 @@ async function submitUpload() { uploading.value = true try { // 浏览器直传 MinIO:presign → PUT(进度条)→ confirm 落库。 - await uploadTutorialPackage(pickedFile.value, pickedFile.value.name, (p) => { + await uploadTutorialPackage(pickedFile.value, pickedFile.value.name, newVersion.value.trim(), (p) => { uploadPercent.value = p uploadMsg.value = p >= 100 ? '上传完成,正在登记教程包...' : `正在上传:${p}%` }) uploadMsg.value = '上传成功,工具台「立即下载教程」已切换为该包' uploadMsgOk.value = true + newVersion.value = '' pickedFile.value = null uploadPercent.value = 0 load() @@ -165,8 +221,13 @@ onMounted(load) 文件名 + + + 大小 - 上传时间 + + + 下载链接 操作 @@ -181,6 +242,10 @@ onMounted(load) {{ row.fileName }} 当前生效 + + {{ row.version }} + + {{ formatFileSize(row.fileSize) }} {{ formatDateTime(row.createdAt) }} @@ -195,10 +260,10 @@ onMounted(load) - 加载中... + 加载中... - {{ keyword ? '暂无匹配教程包' : '暂无教程包记录' }} + {{ keyword ? '暂无匹配教程包' : '暂无教程包记录' }} @@ -209,6 +274,10 @@ onMounted(load)

上传教程 ZIP 包后,工具台首页「立即下载教程」将以下载该包为准(以最新上传的为主)。

+ + +

仅作展示与排序用,可留空;不影响工具台按最新上传下载

+
选择文件 @@ -368,6 +437,46 @@ h3 { text-overflow: ellipsis; vertical-align: middle; } +/* 表头排序:按钮铺满单元格,仅 hover 时加深字色,保持旧版表头观感。 */ +.sort-th { + display: inline-flex; + align-items: center; + gap: 4px; + width: 100%; + padding: 0; + border: 0; + background: none; + color: inherit; + font-family: inherit; + font-size: inherit; + font-weight: inherit; + letter-spacing: inherit; + text-align: left; + cursor: pointer; +} +.sort-th:hover { + color: #2f5d8b; +} +.sort-mark { + color: #8293a5; + font-size: 12px; +} +.sort-th:hover .sort-mark { + color: #5f85ad; +} +.sort-version { + width: 110px; +} +.sort-time { + width: 150px; +} +.version-cell { + display: inline-block; + max-width: 100%; + overflow: hidden; + text-overflow: ellipsis; + vertical-align: middle; +} .tag-active { display: inline-block; margin-left: 6px; diff --git a/admin-frontend-vue/src/pages/records/tutorial-api.ts b/admin-frontend-vue/src/pages/records/tutorial-api.ts index 52daa660..4fb0155a 100644 --- a/admin-frontend-vue/src/pages/records/tutorial-api.ts +++ b/admin-frontend-vue/src/pages/records/tutorial-api.ts @@ -22,11 +22,12 @@ export interface TutorialUploadTarget { objectKey: string uploadUrl: string fileUrl: string + version: string } /** 第一步:向后端申请直传 PUT 预签名地址(登录会话签发,3 分钟有效)。 */ -export async function requestTutorialPresign(fileName: string): Promise { - const { data } = await http.post(TUTORIAL_PRESIGN_ENDPOINT, null, { params: { file_name: fileName } }) +export async function requestTutorialPresign(fileName: string, version = ''): Promise { + const { data } = await http.post(TUTORIAL_PRESIGN_ENDPOINT, null, { params: { file_name: fileName, version } }) const core = (unwrap(data) ?? {}) as Record const uploadUrl = typeof core.upload_url === 'string' ? core.upload_url : '' const objectKey = typeof core.object_key === 'string' ? core.object_key : '' @@ -37,13 +38,14 @@ export async function requestTutorialPresign(fileName: string): Promise { +export async function confirmTutorialPackage(objectKey: string, fileName: string, version = ''): Promise { const { data } = await http.post(TUTORIAL_CONFIRM_ENDPOINT, null, { - params: { object_key: objectKey, file_name: fileName }, + params: { object_key: objectKey, file_name: fileName, version }, }) return parseTutorialPackageUpload(data) } @@ -63,9 +65,10 @@ export async function deleteTutorialPackages(ids: number[]): Promise { export async function uploadTutorialPackage( file: Blob, fileName: string, + version = '', onProgress?: (percent: number) => void, ): Promise { - const target = await requestTutorialPresign(fileName) + const target = await requestTutorialPresign(fileName, version) await directPut.put(target.uploadUrl, file, { headers: { 'Content-Type': 'application/octet-stream' }, onUploadProgress: (event) => { @@ -74,5 +77,5 @@ export async function uploadTutorialPackage( onProgress(total > 0 ? Math.min(Math.round((event.loaded / total) * 100), 100) : 0) }, }) - return confirmTutorialPackage(target.objectKey, fileName) + return confirmTutorialPackage(target.objectKey, fileName, target.version) } diff --git a/admin-frontend-vue/src/pages/records/tutorial-dto.ts b/admin-frontend-vue/src/pages/records/tutorial-dto.ts index 53d039ec..d975dc9f 100644 --- a/admin-frontend-vue/src/pages/records/tutorial-dto.ts +++ b/admin-frontend-vue/src/pages/records/tutorial-dto.ts @@ -4,6 +4,8 @@ export interface TutorialPackageItem { id: number fileName: string + /** 版本号(上传时填写,V126 之前的历史行为空串) */ + version: string objectKey: string fileSize: number fileUrl: string diff --git a/admin-frontend-vue/src/pages/records/tutorial-model.ts b/admin-frontend-vue/src/pages/records/tutorial-model.ts index 3b7b42de..5275cfac 100644 --- a/admin-frontend-vue/src/pages/records/tutorial-model.ts +++ b/admin-frontend-vue/src/pages/records/tutorial-model.ts @@ -19,6 +19,7 @@ export function toTutorialPackageItem(raw: unknown): TutorialPackageItem | null return { id, fileName: text(r.file_name ?? r.fileName), + version: text(r.version), objectKey: text(r.object_key ?? r.objectKey), fileSize: numberOrNull(r.file_size ?? r.fileSize) ?? 0, fileUrl: text(r.file_url ?? r.fileUrl), diff --git a/admin-frontend-vue/tests/align-tutorial.test.ts b/admin-frontend-vue/tests/align-tutorial.test.ts index 9da740aa..a9c1aaf9 100644 --- a/admin-frontend-vue/tests/align-tutorial.test.ts +++ b/admin-frontend-vue/tests/align-tutorial.test.ts @@ -14,16 +14,23 @@ test('align_tutorial_page_registered', () => { test('align_tutorial_page_wiring', () => { const page = readSource('src/pages/records/RecordsTutorialPage.vue') - // 上传入口:选择 zip → 直传(进度条)→ 成功提示留驻弹窗。 + // 上传入口:版本号(可空)→ 选择 zip → 直传(进度条)→ 成功提示留驻弹窗。 assert.match(page, /上传教程包/, '存在上传入口按钮') assert.match(page, /uploadTutorialPackage/, '上传走教程包直传链路') + assert.match(page, /label="版本号"/, '上传弹窗提供版本号输入') + assert.match(page, /newVersion\.value\.trim\(\)/, '版本号去空白后随上传提交') assert.match(page, /请选择 zip 压缩包/, '未选文件时提示') assert.match(page, /仅支持 \.zip 格式/, '格式校验提示') assert.match(page, /上传成功,工具台「立即下载教程」已切换为该包/, '成功提示说明生效位置') assert.match(page, /上传完成,正在登记教程包/, '进度文案') - // 列表:当前生效标记 + 下载 + 删除 + 空态。 + // 列表:版本号列 + 上传时间列 + 当前生效标记 + 下载 + 删除 + 空态。 assert.match(page, /当前生效/, '最新上传的包标记当前生效') assert.match(page, /formatFileSize/, '展示包体大小') + assert.match(page, /row\.version/, '展示版本号列') + assert.match(page, /toggleSort\('version'\)/, '版本号表头可点击排序') + assert.match(page, /toggleSort\('createdAt'\)/, '上传时间表头可点击排序') + assert.match(page, /const sortKey = ref\('createdAt'\)/, '默认排序字段为上传时间') + assert.match(page, /const sortAsc = ref\(false\)/, '默认降序') assert.match(page, /下载/, '操作列提供下载') assert.match(page, /]*download/, '下载链接带 download 强制下载') assert.match(page, /确认删除选中的/, '批量删除二次确认') @@ -37,7 +44,8 @@ test('align_tutorial_api_contract', () => { assert.match(api, /\$\{TUTORIAL_UPLOAD_ENDPOINT\}\/presign/, '直传预签名端点') assert.match(api, /\$\{TUTORIAL_UPLOAD_ENDPOINT\}\/confirm/, '确认端点') assert.match(api, /\$\{TUTORIAL_UPLOAD_ENDPOINT\}\/delete/, '删除端点') - assert.match(api, /params: \{ object_key: objectKey, file_name: fileName \}/, '确认回传对象 key 与文件名') + assert.match(api, /params: \{ file_name: fileName, version \}/, '预签名回传文件名与版本号') + assert.match(api, /params: \{ object_key: objectKey, file_name: fileName, version \}/, '确认回传对象 key、文件名与版本号') assert.match(api, /withCredentials: false/, '直传实例不挂会话拦截器') }) @@ -46,4 +54,5 @@ test('align_tutorial_model_parsers', () => { assert.match(model, /r\.file_name \?\? r\.fileName/, '兼容 snake/camel 文件名') assert.match(model, /r\.file_url \?\? r\.fileUrl/, '兼容 snake/camel 下载链接') assert.match(model, /r\.file_size \?\? r\.fileSize/, '兼容 snake/camel 文件大小') + assert.match(model, /version: text\(r\.version\)/, '解析版本号(缺失为空串,兼容历史行)') }) diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java index cbd56877..8a524524 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java @@ -30,15 +30,24 @@ public class AppearancePatentProperties { private int llmFirstAttemptReadTimeoutMillis = 60000; private int llmBatchSize = 10; /** - * 批内行级并发数,默认等于批量大小 + * 批内行级并发数。批次串行提交,每行串行发 2 个 LLM 请求, + * 故该值≈单任务对 LLM 网关的瞬时并发;默认与品牌检测同为 5,避免多任务并行时成倍放大。 */ - private int llmRowConcurrency = 10; + private int llmRowConcurrency = 5; /** * 每行每个 LLM 请求的重试次数(含首次) */ private int llmRetryTimes = 3; private int staleTimeoutMinutes = 30; private String staleFinalizeCron = "0 */2 * * * *"; + + /** + * 二次判死线(心跳正常但连续 N 分钟无结果分片上报)阈值,分钟;<=0 关闭本线。 + * 既有判定以 Redis heartbeat stale 为主信号,但心跳随 Python HTTP 心跳每分钟刷新—— + * 主线程卡死时心跳线程照发,任务永远不 stale(与生产 28131 同型缺口)。 + * 本线改看 biz_task_scope_state.last_chunk_at(仅分片上传时刷新)。 + */ + private int noResultUploadTimeoutMinutes = 180; /** * 末尾不足一批的数据等待该时长后强制提交检测。 * Python 回传超时后会直接封口;这个配置处理 Python 慢回传但仍未超时的零头批次。 diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java index f81b8cdb..b07e70c6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java @@ -10,10 +10,19 @@ public class BrandCheckProperties { private String path = "/brand_check"; private String token = ""; private String defaultStrategy = "Terms"; - /** 单品牌商标查询的重试次数(含首次),连续失败超过该次数才判定为查询失败。 */ - private int retryTimes = 3; - /** 每次查询失败后到下一次重试前的等待毫秒数。 */ + /** + * 单品牌商标查询的重试次数(含首次),连续失败超过该次数才判定为查询失败。 + * 原为 3:16890 偶发限流几秒内即恢复,3 次(前两次间隔各 1s)恢复不了就把结论 + * 写成「查询失败」,对客户是硬伤;2026-09-14 与客户端品牌一致提到 10 次。 + */ + private int retryTimes = 10; + /** 每次查询失败后到下一次重试前的等待毫秒数(基准值,按重试轮次递增)。 */ private int retryIntervalMillis = 1000; + /** + * 单次重试等待的上限毫秒数。等待按 retryIntervalMillis × 第几次重试 递增后封顶, + * 10 次尝试合计约 45s,既能覆盖限流窗口又不会让单个品牌长时间占住查询线程。 + */ + private int retryMaxIntervalMillis = 10000; private int connectTimeoutMillis = 10000; private int readTimeoutMillis = 60000; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/BrandProgressProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/BrandProgressProperties.java index c2700062..41e0d0f2 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/BrandProgressProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/BrandProgressProperties.java @@ -10,4 +10,11 @@ public class BrandProgressProperties { private long failedTtlHours = 2; private long heartbeatTimeoutMinutes = 15; private String staleCheckCron = "0 */2 * * * *"; + + /** + * 二次判死线(心跳正常但连续 N 分钟无结果上报)阈值,分钟;<=0 关闭本线。 + * 既有心跳线以 updated_at/last_heartbeat_at 陈旧为判据,而前端心跳会持续刷新它们—— + * 主线程卡死时心跳线程照发,任务永远命不中。本线改看结果上报时写入的 last_result_at。 + */ + private long noResultUploadTimeoutMinutes = 180; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java index 80622f56..37f56484 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java @@ -45,6 +45,19 @@ public class DeleteBrandProgressProperties { */ private long withdrawStaleTimeoutMinutes = 30; + /** + * 「心跳正常但连续 N 分钟无结果分片上报」的二次判死阈值(分钟),默认 3 小时。 + * + *

既有各模块心跳线的候选条件是 updated_at 陈旧,而客户端心跳会持续刷新 updated_at—— + * 主线程卡死时心跳线程照发,任务永远命不中(生产 28131 卡死 12h+ 仍 RUNNING)。 + * 本线改用 biz_task_scope_state.last_chunk_at(只随结果分片上报刷新)作判据, + * 从未上报(无 scope 行或全 NULL)的任务跳过,避免误杀首批较慢的正常任务。 + */ + private long noResultUploadTimeoutMinutes = 180; + + /** 二次判死线开关:false = 整段不扫描(观察期与回滚用,改环境变量即生效)。 */ + private boolean noResultUploadCheckEnabled = true; + /** * 组装结果文件时允许缺失的最大分片数:缺失不超过该值时降级组装(缺失行状态写"未回传"), * 保证任务不因少量分片丢失整体白跑;超出时任务失败,但仍会在重试耗尽后尝试 diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java index be573e7c..47a038be 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java @@ -23,6 +23,14 @@ public class SimilarAsinProperties { private int staleTimeoutMinutes = 30; private String staleFinalizeCron = "0 */2 * * * *"; + /** + * 二次判死线(心跳正常但连续 N 分钟无结果分片上报)阈值,分钟;<=0 关闭本线。 + * 既有判定以 Redis heartbeat stale 为主信号,但心跳随 Python HTTP 心跳每分钟刷新—— + * 主线程卡死时心跳线程照发,任务永远不 stale(与生产 28131 同型缺口)。 + * 本线改看 biz_task_scope_state.last_chunk_at(仅分片上传时刷新)。 + */ + private int noResultUploadTimeoutMinutes = 180; + /** * 末尾零头 batch 的强制 flush 阈值(分钟):当不足 llmBatchSize 的零头 row * 长时间挂着(Python 慢回传)时触发提交。 diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index 7675e268..ac91a970 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -40,6 +40,7 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; import com.nanri.aiimage.modules.task.model.entity.FileResultEntity; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; @@ -82,6 +83,7 @@ import java.util.ArrayList; import java.util.Comparator; import java.util.LinkedHashMap; import java.util.LinkedHashSet; +import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; @@ -601,6 +603,7 @@ public class AppearancePatentTaskService { } } } + finalizeNoUploadStaleTasks(); } public void debugFinalizeStaleTask(Long taskId) { @@ -661,6 +664,64 @@ public class AppearancePatentTaskService { return updatedMillis <= thresholdMillis; } + /** + * 二次判死:Python 心跳正常(Redis heartbeat 新鲜)但连续 N 分钟无结果分片上报。 + * + *

既有判定以 Redis heartbeat 为 stale 主信号(P1-7),但该心跳随 Python 的 HTTP 心跳 + * 每分钟刷新——主线程卡死时心跳线程照发,任务永远不 stale(与生产 28131 同型缺口)。 + * 本线改看 biz_task_scope_state.last_chunk_at(仅在 persistSubmittedChunk 上传分片时刷新), + * 命中后走既有 finalizeStaleTask(封口上传 + LLM 收尾,不粗暴杀任务)。 + * 从未上报(无 scope 行或全 NULL)的任务跳过,避免误杀首批较慢的正常任务。 + */ + private void finalizeNoUploadStaleTasks() { + long minutes = properties.getNoResultUploadTimeoutMinutes(); + if (minutes <= 0) { + return; + } + List tasks = listStaleFinalizeCandidates(); + if (tasks.isEmpty()) { + return; + } + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes); + long heartbeatThresholdMillis = LocalDateTime.now() + .minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())) + .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); + Map lastResultAtByTaskId = new HashMap<>(); + for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds( + tasks.stream().map(FileTaskEntity::getId).toList())) { + if (dto.taskId() != null && dto.lastChunkAt() != null) { + lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt()); + } + } + for (FileTaskEntity task : tasks) { + if (isHeartbeatStale(task, heartbeatThresholdMillis)) { + // 心跳已 stale:归既有心跳线处理 + continue; + } + LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId()); + if (lastResultAt == null || lastResultAt.isAfter(cutoff)) { + continue; + } + TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(task.getId(), 0L); + if (lockHandle == null) { + continue; + } + try (lockHandle) { + log.warn("[appearance-patent] 心跳正常但 {} 分钟无结果分片上报,按卡死收尾 taskId={} lastResultAt={}", + minutes, task.getId(), lastResultAt); + String error = "Python heartbeat alive but no result chunk uploaded for " + minutes + " minutes"; + if (transactionManager != null) { + inNewTransaction(() -> { + finalizeStaleTask(task.getId(), error); + return null; + }); + } else { + finalizeStaleTask(task.getId(), error); + } + } + } + } + /** * /result 提交的纯计算阶段(无事务注解):读任务 + 校验归属 + 分组展开 + * 序列化 + payload 存储 + 哈希预计算,全部在事务开始前完成。 diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java index e3f2090d..d6c05fdb 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java @@ -121,9 +121,10 @@ public class BrandCheckClient { lastFailure = ex; response = null; if (attempt < attempts) { - log.warn("[brand-check] 请求异常将重试 brand={} attempt={}/{} err={}", - brand, attempt, attempts, ex.getMessage()); - sleepBeforeRetry(); + long retryDelayMillis = retryDelayMillis(attempt); + log.warn("[brand-check] 请求异常将重试 brand={} attempt={}/{} 等待={}ms err={}", + brand, attempt, attempts, retryDelayMillis, ex.getMessage()); + sleepBeforeRetry(retryDelayMillis); } continue; } @@ -132,9 +133,10 @@ public class BrandCheckClient { return new BrandCheckOutcome(nullToEmpty(response.getFaildData()), List.of()); } if (attempt < attempts) { - log.warn("[brand-check] 服务端返回查询失败将重试 brand={} attempt={}/{}", - brand, attempt, attempts); - sleepBeforeRetry(); + long retryDelayMillis = retryDelayMillis(attempt); + log.warn("[brand-check] 服务端返回查询失败将重试 brand={} attempt={}/{} 等待={}ms", + brand, attempt, attempts, retryDelayMillis); + sleepBeforeRetry(retryDelayMillis); } } log.warn("[brand-check] 重试耗尽判定查询失败 brand={} attempts={} lastErr={}", @@ -144,8 +146,18 @@ public class BrandCheckClient { response == null ? List.of(brand) : nullToEmpty(response.getQueryFaildData())); } - private void sleepBeforeRetry() { - long delayMillis = Math.max(0L, properties.getRetryIntervalMillis()); + /** + * 第 attempt 次重试前的等待毫秒数:按基准间隔随轮次递增后封顶。 + * 限流窗口通常只有几秒,固定 1s 间隔反复打过去救不回来;递增等待能覆盖窗口, + * 封顶则保证单个品牌不会长时间占住查询线程(并发度只有 3)。 + */ + private long retryDelayMillis(int attempt) { + long base = Math.max(0L, properties.getRetryIntervalMillis()); + long cap = Math.max(base, properties.getRetryMaxIntervalMillis()); + return Math.min(base * Math.max(1, attempt), cap); + } + + private void sleepBeforeRetry(long delayMillis) { if (delayMillis <= 0L) { return; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java index f5d66391..0f820e5d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java @@ -60,6 +60,8 @@ public class BrandTaskProgressCacheService { values.put("finished_files", String.valueOf(Math.max(finishedFiles, 0))); values.put("updated_at", now); values.put("last_heartbeat_at", now); + // 结果上报专属信号(二次判死线用):心跳/touchHeartbeat 不写它,只有结果回传才刷新 + values.put("last_result_at", now); try { stringRedisTemplate.opsForHash().putAll(key, values); stringRedisTemplate.expire(key, ttl()); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java index fc69220b..92ccabc1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java @@ -1537,9 +1537,58 @@ public class BrandTaskService { failStaleRunningTask(task.getId(), "前端长时间无响应,任务已自动失败"); } } + failNoUploadStaleRunningTasks(); } } + /** + * 二次判死:前端心跳正常但连续 N 分钟无结果上报(治「心跳续命」的假活任务)。 + * + *

既有心跳线候选条件是 updated_at/last_heartbeat_at 陈旧,而前端心跳会持续刷新它们—— + * 主线程卡死时心跳线程照发,任务永远命不中(与生产 28131 同型缺口)。 + * 本线候选取「心跳新鲜 + 创建超过 N 分钟」,判据用结果上报时写入 progress hash 的 + * last_result_at(仅 saveProgressFromResult 写);从未上报(无该字段)跳过,避免误杀首批较慢的正常任务。 + */ + private void failNoUploadStaleRunningTasks() { + long minutes = brandProgressProperties.getNoResultUploadTimeoutMinutes(); + if (minutes <= 0) { + return; + } + LocalDateTime heartbeatThreshold = LocalDateTime.now() + .minusMinutes(brandProgressProperties.getHeartbeatTimeoutMinutes()); + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes); + List runningTasks = brandCrawlTaskMapper.selectList(new LambdaQueryWrapper() + .eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING) + .ge(BrandCrawlTaskEntity::getUpdatedAt, heartbeatThreshold) + .lt(BrandCrawlTaskEntity::getCreatedAt, cutoff)); + for (BrandCrawlTaskEntity task : runningTasks) { + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(task.getId(), 0L); + if (taskLockHandle == null) { + continue; + } + try (taskLockHandle) { + Map progress = brandTaskProgressCacheService.getProgress(task.getId()); + long lastResultAt = 0L; + try { + lastResultAt = Long.parseLong(String.valueOf(progress.getOrDefault("last_result_at", "0"))); + } catch (Exception ignored) { + } + if (lastResultAt <= 0L) { + // 从未上报结果:保守跳过(首批可能较慢) + continue; + } + LocalDateTime lastResult = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastResultAt), ZoneId.systemDefault()); + if (lastResult.isAfter(cutoff)) { + continue; + } + log.warn("[brand-stale-check] no-upload failing taskId={} lastResultAt={} timeoutMinutes={}", + task.getId(), lastResult, minutes); + failStaleRunningTask(task.getId(), + "连续 " + minutes + " 分钟无结果回传,疑似客户端任务卡死,任务已自动失败(最后结果上报 " + lastResult + ")"); + } + } + } + private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) { TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, waitMillis); if (lockHandle == null) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index c79e3e30..46306797 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -51,6 +51,7 @@ import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; import com.nanri.aiimage.modules.task.mapper.TaskResultItemMapper; import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; import com.nanri.aiimage.modules.task.model.entity.FileResultEntity; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; @@ -179,6 +180,10 @@ public class CollectDataService { @Value("${aiimage.collect-data.stale-timeout-minutes:30}") private long staleTimeoutMinutes; + /** 二次判死线(心跳正常但连续 N 分钟无结果分片上报)阈值,分钟;<=0 关闭本线。 */ + @Value("${aiimage.collect-data.no-result-upload-timeout-minutes:180}") + private long noResultUploadTimeoutMinutes; + @Value("${aiimage.collect-data.max-source-file-bytes:0}") private Long maxSourceFileBytes; @@ -526,6 +531,7 @@ public class CollectDataService { for (FileTaskEntity task : tasks) { finalizeStaleTask(task.getId(), threshold, timeoutMinutes); } + finalizeNoUploadStaleTasks(threshold); } } @@ -565,6 +571,86 @@ public class CollectDataService { } } + /** + * 二次判死:心跳正常但连续 N 分钟无结果分片上报(治「心跳续命」的假活任务)。 + * + *

既有心跳线候选条件是 updated_at 陈旧,而客户端心跳会持续刷新 updated_at—— + * 主线程卡死(浏览器自动化等待/异常)时心跳线程照发,任务永远命不中。 + * 本线候选取「心跳新鲜(既有线放过)+ 创建超过 N 分钟」,判据用 + * biz_task_scope_state.last_chunk_at(只随结果分片上报刷新); + * 从未上报(无 scope 行或全 NULL)的任务跳过,避免误杀首批较慢的正常任务。 + */ + private void finalizeNoUploadStaleTasks(LocalDateTime heartbeatThreshold) { + long minutes = noResultUploadTimeoutMinutes; + if (minutes <= 0) { + return; + } + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes); + List candidates = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, STATUS_RUNNING) + .ge(FileTaskEntity::getUpdatedAt, heartbeatThreshold) + .lt(FileTaskEntity::getCreatedAt, cutoff) + .orderByAsc(FileTaskEntity::getCreatedAt) + .last("limit 200")); + if (candidates.isEmpty()) { + return; + } + Map lastResultAtByTaskId = new HashMap<>(); + for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds( + candidates.stream().map(FileTaskEntity::getId).toList())) { + if (dto.taskId() != null && dto.lastChunkAt() != null) { + lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt()); + } + } + for (FileTaskEntity task : candidates) { + LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId()); + if (lastResultAt == null || lastResultAt.isAfter(cutoff)) { + continue; + } + finalizeNoUploadStaleTask(task.getId(), lastResultAt, minutes); + } + } + + /** 单个任务的二次判死收尾:锁内复查心跳后,有分片→组装部分工作簿;无分片→标失败。 */ + private void finalizeNoUploadStaleTask(Long taskId, LocalDateTime lastResultAt, long minutes) { + try (TaskDistributedLockService.LockHandle lock = + taskDistributedLockService.acquire(MODULE_TYPE, taskId, 0L)) { + if (lock == null) { + return; + } + FileTaskEntity task = fileTaskMapper.selectById(taskId); + LocalDateTime heartbeatFreshAfter = LocalDateTime.now().minusMinutes(Math.max(1L, staleTimeoutMinutes)); + if (task == null + || !MODULE_TYPE.equals(task.getModuleType()) + || !STATUS_RUNNING.equals(task.getStatus()) + || task.getUpdatedAt() == null + || task.getUpdatedAt().isBefore(heartbeatFreshAfter)) { + // 已终态、或心跳在排队期间回落到陈旧(交回既有心跳线处理) + return; + } + if (taskFileJobService.countUnfinishedAssembleJobs(taskId, MODULE_TYPE) > 0L) { + return; + } + FileResultEntity result = ensureTaskResult(task); + CollectDataStats stats = loadStats(task); + if (hasReceivedChunks(taskId)) { + enqueueFinalWorkbook(task, result, stats); + log.warn("[collect-data] no-upload stale task enqueued partial workbook taskId={} lastResultAt={} timeoutMinutes={} finalRows={}", + taskId, lastResultAt, minutes, stats.finalRowCount); + return; + } + markTaskFailed(task, result, + "连续 " + minutes + " 分钟无结果回传,疑似客户端任务卡死,任务已自动失败(最后结果上报 " + lastResultAt + ")", + stats); + log.warn("[collect-data] no-upload stale task failed without chunks taskId={} lastResultAt={} timeoutMinutes={}", + taskId, lastResultAt, minutes); + } catch (Exception ex) { + log.warn("[collect-data] no-upload stale task finalization failed taskId={} msg={}", + taskId, ex.getMessage(), ex); + } + } + public CollectDataDashboardVo dashboard(Long userId) { CollectDataDashboardVo vo = new CollectDataDashboardVo(); vo.setPendingTaskCount(countActiveTasks(userId)); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java index d38eca1a..47cdb5b8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java @@ -21,9 +21,12 @@ import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskService; import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskCacheService; import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskService; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskHeartbeatPositionService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; @@ -55,6 +58,10 @@ public class DeleteBrandStaleTaskService { private static final String MODULE_TYPE_PATROL_DELETE = "PATROL_DELETE"; private static final String MODULE_TYPE_QUERY_ASIN = "QUERY_ASIN"; private static final String MODULE_TYPE_WITHDRAW = "WITHDRAW"; + /** 二次判死线(心跳正常但无结果上报)覆盖的模块:走结果分片上报表、last_chunk_at 有效的模块。 */ + private static final List NO_RESULT_UPLOAD_CHECKED_MODULES = List.of( + MODULE_TYPE_DELETE_BRAND, MODULE_TYPE_PRODUCT_RISK, MODULE_TYPE_PRICE_TRACK, + MODULE_TYPE_SHOP_MATCH, MODULE_TYPE_PATROL_DELETE, MODULE_TYPE_QUERY_ASIN, MODULE_TYPE_WITHDRAW); private static final Duration STALE_CHECK_LOCK_TTL = Duration.ofMinutes(10); private static final Duration FINALIZE_CHECK_LOCK_TTL = Duration.ofMinutes(10); private static final Duration TEMP_DIR_CLEANUP_LOCK_TTL = Duration.ofMinutes(10); @@ -83,6 +90,8 @@ public class DeleteBrandStaleTaskService { private final TaskDistributedLockService taskDistributedLockService; private final TaskFileJobService taskFileJobService; private final ShopDataCrawlTaskService shopDataCrawlTaskService; + private final TaskScopeStateMapper taskScopeStateMapper; + private final TaskHeartbeatPositionService taskHeartbeatPositionService; @Value("${aiimage.temp-dir.retention-hours:24}") private long tempDirRetentionHours; @@ -104,19 +113,21 @@ public class DeleteBrandStaleTaskService { ShopMatchStaleCheckStats patrolDeleteStats = failStalePatrolDeleteTasks(); ShopMatchStaleCheckStats queryAsinStats = failStaleQueryAsinTasks(); ShopMatchStaleCheckStats withdrawStats = failStaleWithdrawTasks(); + NoResultUploadStaleCheckStats noUploadStats = failNoResultUploadTasks(); for (Map.Entry delegated : delegatedStaleChecks().entrySet()) { runModuleStaleCheck(delegated.getKey(), delegated.getValue()); } // 周期每 2 分钟一轮,各模块 summary 合并为单行,避免定期刷屏 // 注意:每段占位符数量必须与实参一致——此前每段 5 个占位符只传 4 个参数, // 导致 withdraw 之后的取值整体错位、末尾 elapsedMs/thread 打成字面量 - log.info("[stale-check] summary product-risk(s={} f={} x={} p={}) price-track(s={} f={} x={} p={}) shop-match(s={} f={} x={} p={}) patrol-delete(s={} f={} x={} p={}) query-asin(s={} f={} x={} p={}) withdraw(s={} f={} x={} p={}) elapsedMs={} thread={}", + log.info("[stale-check] summary product-risk(s={} f={} x={} p={}) price-track(s={} f={} x={} p={}) shop-match(s={} f={} x={} p={}) patrol-delete(s={} f={} x={} p={}) query-asin(s={} f={} x={} p={}) withdraw(s={} f={} x={} p={}) no-upload(c={} f={} x={}) elapsedMs={} thread={}", stats.scannedTaskCount, stats.finalizedTaskCount, stats.failedTaskCount, stats.skippedTaskCount, priceTrackStats.scannedTaskCount, priceTrackStats.finalizedTaskCount, priceTrackStats.failedTaskCount, priceTrackStats.skippedTaskCount, shopMatchStats.scannedTaskCount, shopMatchStats.finalizedTaskCount, shopMatchStats.failedTaskCount, shopMatchStats.skippedTaskCount, patrolDeleteStats.scannedTaskCount, patrolDeleteStats.finalizedTaskCount, patrolDeleteStats.failedTaskCount, patrolDeleteStats.skippedTaskCount, queryAsinStats.scannedTaskCount, queryAsinStats.finalizedTaskCount, queryAsinStats.failedTaskCount, queryAsinStats.skippedTaskCount, withdrawStats.scannedTaskCount, withdrawStats.finalizedTaskCount, withdrawStats.failedTaskCount, withdrawStats.skippedTaskCount, + noUploadStats.scannedTaskCount, noUploadStats.failedTaskCount, noUploadStats.skippedTaskCount, System.currentTimeMillis() - startedAt, Thread.currentThread().getName()); } @@ -711,6 +722,125 @@ public class DeleteBrandStaleTaskService { return stats; } + /** + * 二次判死:心跳正常但连续 N 分钟无结果分片上报(治「心跳续命」的假活任务)。 + * + *

既有各模块心跳线的候选条件都是 updated_at 陈旧,而客户端心跳会持续刷新 updated_at—— + * 主线程卡死(浏览器自动化等待/异常)时心跳线程照发,任务永远命不中(生产 28131 卡死 12h+ 仍 RUNNING)。 + * 本线改用 biz_task_scope_state.last_chunk_at(只随结果分片上报刷新)作判据; + * 无 scope 行或 last_chunk_at 全 NULL(从未上报)的任务跳过,避免误杀首批较慢的正常任务。 + */ + private NoResultUploadStaleCheckStats failNoResultUploadTasks() { + NoResultUploadStaleCheckStats stats = new NoResultUploadStaleCheckStats(); + if (!deleteBrandProgressProperties.isNoResultUploadCheckEnabled()) { + return stats; + } + long minutes = Math.max(1L, deleteBrandProgressProperties.getNoResultUploadTimeoutMinutes()); + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes); + List candidates = fileTaskMapper.selectList(new LambdaQueryWrapper() + .select(FileTaskEntity::getId, FileTaskEntity::getModuleType, + FileTaskEntity::getCreatedAt, FileTaskEntity::getUpdatedAt) + .eq(FileTaskEntity::getStatus, "RUNNING") + .in(FileTaskEntity::getModuleType, NO_RESULT_UPLOAD_CHECKED_MODULES) + .lt(FileTaskEntity::getCreatedAt, cutoff) + .orderByAsc(FileTaskEntity::getCreatedAt) + .last("limit 200")); + stats.scannedTaskCount = candidates.size(); + if (candidates.isEmpty()) { + return stats; + } + Map lastResultAtByTaskId = taskScopeStateMapper + .selectLastChunkAtByTaskIds(candidates.stream().map(FileTaskEntity::getId).toList()) + .stream() + .filter(dto -> dto.taskId() != null && dto.lastChunkAt() != null) + .collect(Collectors.toMap(TaskScopeLastChunkDto::taskId, TaskScopeLastChunkDto::lastChunkAt, (a, b) -> a)); + for (FileTaskEntity task : candidates) { + LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId()); + // 从未上报(无 scope 行或全 NULL)或上报仍新鲜:正常推进,不必处理 + if (lastResultAt == null || lastResultAt.isAfter(cutoff)) { + continue; + } + String moduleType = task.getModuleType(); + if (hasPendingAssembleJobs(task.getId(), moduleType)) { + stats.skippedTaskCount++; + log.info("[stale-check] no-upload skip pending-assemble-jobs taskId={} moduleType={}", task.getId(), moduleType); + continue; + } + TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(moduleType, task.getId()); + if (taskLockHandle == null) { + stats.skippedTaskCount++; + continue; + } + try (taskLockHandle) { + try { + tryFinalizeNoResultUploadTask(moduleType, task.getId()); + } catch (Exception ex) { + // finalize 抛异常(组装失败等)不代表任务活着:继续走 CAS 判死 + log.warn("[stale-check] no-upload finalize threw taskId={} moduleType={} msg={}", task.getId(), moduleType, ex.getMessage()); + } + // finalize 已终结的任务 CAS 自然不命中(状态不再是 RUNNING),无需回读 + String lastPosition = taskHeartbeatPositionService.describe(task.getId()); + int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(FileTaskEntity::getId, task.getId()) + .eq(FileTaskEntity::getModuleType, moduleType) + .eq(FileTaskEntity::getStatus, "RUNNING") + .set(FileTaskEntity::getStatus, "FAILED") + .set(FileTaskEntity::getErrorMessage, buildNoResultUploadFailReason(minutes, lastResultAt, lastPosition)) + .set(FileTaskEntity::getUpdatedAt, LocalDateTime.now()) + .set(FileTaskEntity::getFinishedAt, LocalDateTime.now())); + if (updated > 0) { + stats.failedTaskCount++; + deleteNoResultUploadTaskCache(moduleType, task.getId()); + log.warn("[stale-check] no-upload failed taskId={} moduleType={} lastResultAt={} taskUpdatedAt={} timeoutMinutes={}", + task.getId(), moduleType, lastResultAt, task.getUpdatedAt(), minutes); + } else { + stats.skippedTaskCount++; + log.info("[stale-check] no-upload task already finalized by compensation taskId={} moduleType={}", task.getId(), moduleType); + } + } + } + return stats; + } + + /** 按模块调用各自的收尾入口(尽力组装部分结果;已终结的任务由后续 CAS 自然放行)。 */ + private void tryFinalizeNoResultUploadTask(String moduleType, Long taskId) { + switch (moduleType) { + case MODULE_TYPE_DELETE_BRAND -> deleteBrandRunService.tryFinalizeTask(taskId, true); + case MODULE_TYPE_PRODUCT_RISK -> productRiskTaskService.tryFinalizeTask(taskId, true); + case MODULE_TYPE_PRICE_TRACK -> priceTrackTaskService.tryFinalizeTask(taskId, true); + case MODULE_TYPE_SHOP_MATCH -> shopMatchTaskService.tryFinalizeTask(taskId, true); + case MODULE_TYPE_PATROL_DELETE -> patrolDeleteTaskService.tryFinalizeTask(taskId, true); + case MODULE_TYPE_QUERY_ASIN -> queryAsinTaskService.tryFinalizeTask(taskId, true); + case MODULE_TYPE_WITHDRAW -> withdrawTaskService.tryFinalizeTask(taskId, true); + default -> { } + } + } + + /** 按模块删除各自缓存(与既有各段 CAS 翻转后的清理一致)。 */ + private void deleteNoResultUploadTaskCache(String moduleType, Long taskId) { + switch (moduleType) { + case MODULE_TYPE_DELETE_BRAND -> deleteBrandTaskCacheService.delete(taskId); + case MODULE_TYPE_PRODUCT_RISK -> productRiskTaskCacheService.deleteTaskCache(taskId); + case MODULE_TYPE_PRICE_TRACK -> priceTrackTaskCacheService.deleteTaskCache(taskId); + case MODULE_TYPE_SHOP_MATCH -> shopMatchTaskCacheService.deleteTaskCache(taskId); + case MODULE_TYPE_PATROL_DELETE -> patrolDeleteTaskCacheService.deleteTaskCache(taskId); + case MODULE_TYPE_QUERY_ASIN -> queryAsinTaskCacheService.deleteTaskCache(taskId); + case MODULE_TYPE_WITHDRAW -> withdrawTaskCacheService.deleteTaskCache(taskId); + default -> { } + } + } + + /** 二次判死文案:写明最后结果上报时间与最后处理位置,便于区分「卡死」与「长间隔」并定位卡点。 */ + private String buildNoResultUploadFailReason(long minutes, LocalDateTime lastResultAt, String position) { + StringBuilder sb = new StringBuilder("连续 ").append(minutes) + .append(" 分钟无结果回传,疑似客户端任务卡死,任务已自动失败(最后结果上报 ").append(lastResultAt); + if (position != null && !position.isBlank()) { + sb.append(",最后处理位置:").append(position); + } + sb.append(")"); + return sb.toString(); + } + private TaskDistributedLockService.LockHandle acquireTaskLock(String moduleType, Long taskId) { TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(moduleType, taskId, 0L); if (lockHandle == null) { @@ -849,4 +979,10 @@ public class DeleteBrandStaleTaskService { private int failedTaskCount; private int skippedTaskCount; } + + private static final class NoResultUploadStaleCheckStats { + private int scannedTaskCount; + private int failedTaskCount; + private int skippedTaskCount; + } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java index e819e3b9..19854a2e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java @@ -36,6 +36,7 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest; import com.nanri.aiimage.modules.task.model.entity.FileResultEntity; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; @@ -70,6 +71,7 @@ import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.function.Function; +import java.util.function.Predicate; import java.util.stream.Collectors; import java.util.stream.Collectors; import com.nanri.aiimage.modules.task.model.vo.TaskProgressLightBatchVo; @@ -116,6 +118,10 @@ public class PublishTaskService { @Value("${aiimage.publish.stale-timeout-minutes:30}") private int staleTimeoutMinutes; + /** 二次判死线(心跳正常但连续 N 分钟无结果分片上报)阈值,分钟;<=0 关闭本线。 */ + @Value("${aiimage.publish.no-result-upload-timeout-minutes:180}") + private int noResultUploadTimeoutMinutes; + public PublishParseVo parseAndCreateTask(PublishParseRequest request) { validateParseRequest(request); List preparedFiles = new ArrayList<>(); @@ -459,6 +465,7 @@ public class PublishTaskService { candidate.getId(), ex.getMessage()); } } + failNoUploadStaleTasks(); } } @@ -685,10 +692,93 @@ public class PublishTaskService { } private void failStaleTaskLocked(Long taskId, LocalDateTime threshold) { + failStaleTaskLockedInternal(taskId, + task -> task.getUpdatedAt() != null && task.getUpdatedAt().isBefore(threshold), + "任务心跳超时"); + } + + /** + * 二次判死:Python 心跳正常但连续 N 分钟无结果分片上报(治「心跳续命」的假活任务)。 + * + *

既有心跳线候选条件是 updated_at 陈旧,而客户端心跳会持续刷新 updated_at—— + * 主线程卡死时心跳线程照发,任务永远命不中(与生产 28131 同型缺口)。 + * owner 过滤与 A 线保持一致(避免双节点同时处理非本实例任务); + * 判据用 biz_task_scope_state.last_chunk_at(仅分片上传时刷新);从未上报跳过,避免误杀首批较慢的正常任务。 + */ + private void failNoUploadStaleTasks() { + long minutes = noResultUploadTimeoutMinutes; + if (minutes <= 0) { + return; + } + LocalDateTime heartbeatThreshold = LocalDateTime.now().minusMinutes(Math.max(5, staleTimeoutMinutes)); + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes); + List candidates = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, STATUS_RUNNING) + .and(owner -> owner + .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) IS NULL") + .or() + .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = ''") + .or() + .apply("JSON_UNQUOTE(JSON_EXTRACT(result_json, '$.ownerInstanceId')) = {0}", currentInstanceId())) + .ge(FileTaskEntity::getUpdatedAt, heartbeatThreshold) + .lt(FileTaskEntity::getCreatedAt, cutoff) + .orderByAsc(FileTaskEntity::getCreatedAt) + .last("limit 100")); + if (candidates.isEmpty()) { + return; + } + Map lastResultAtByTaskId = new LinkedHashMap<>(); + for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds( + candidates.stream().map(FileTaskEntity::getId).toList())) { + if (dto.taskId() != null && dto.lastChunkAt() != null) { + lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt()); + } + } + for (FileTaskEntity candidate : candidates) { + LocalDateTime lastResultAt = lastResultAtByTaskId.get(candidate.getId()); + if (lastResultAt == null || lastResultAt.isAfter(cutoff)) { + continue; + } + if (taskFileJobService.countUnfinishedAssembleJobs(candidate.getId(), MODULE_TYPE) > 0L) { + continue; + } + try (TaskDistributedLockService.LockHandle lock = + taskDistributedLockService.acquire(MODULE_TYPE, candidate.getId(), 0L)) { + if (lock == null) { + continue; + } + String error = "连续 " + minutes + " 分钟无结果回传,疑似客户端任务卡死,任务已自动失败(最后结果上报 " + lastResultAt + ")"; + transactionTemplate.executeWithoutResult(status -> failStaleTaskLockedInternal( + candidate.getId(), + task -> { + // 排队期间复查:心跳回落陈旧 → 交回 A 线;出现新结果上报 → 任务已恢复 + if (task.getUpdatedAt() == null || task.getUpdatedAt().isBefore(heartbeatThreshold)) { + return false; + } + LocalDateTime freshLastChunk = queryLastChunkAt(task.getId()); + return freshLastChunk != null && !freshLastChunk.isAfter(cutoff); + }, + error)); + } catch (Exception ex) { + log.warn("[publish] no-upload stale task cleanup failed taskId={} msg={}", + candidate.getId(), ex.getMessage()); + } + } + } + + /** 单任务最后一次结果分片上报时间;无记录返回 null。 */ + private LocalDateTime queryLastChunkAt(Long taskId) { + List rows = taskScopeStateMapper.selectLastChunkAtByTaskIds(List.of(taskId)); + return rows.isEmpty() ? null : rows.get(0).lastChunkAt(); + } + + /** 收尾条件由调用方判定(A 线=心跳陈旧;B 线=心跳新鲜但结果上报陈旧),条件通过后走同一收尾路径。 */ + private void failStaleTaskLockedInternal(Long taskId, Predicate staleCheck, String error) { FileTaskEntity task = fileTaskMapper.selectById(taskId); if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !STATUS_RUNNING.equals(task.getStatus()) - || task.getUpdatedAt() == null || !task.getUpdatedAt().isBefore(threshold)) { + || !staleCheck.test(task)) { return; } ensureTaskOwnedByCurrentInstance(task, "cleanup stale publish task"); @@ -696,7 +786,6 @@ public class PublishTaskService { return; } LocalDateTime now = LocalDateTime.now(); - String error = "任务心跳超时"; List files = listTaskFiles(taskId); for (PublishFileEntity file : files) { if (!STATUS_PENDING.equals(file.getStatus()) && !STATUS_RUNNING.equals(file.getStatus())) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java index 8a17756d..6dca797f 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java @@ -32,6 +32,7 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; import com.nanri.aiimage.modules.task.model.entity.FileResultEntity; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; @@ -126,6 +127,10 @@ public class ShopDataCrawlTaskService { @Value("${aiimage.shop-data-crawl.stale-timeout-minutes:30}") private long staleTimeoutMinutes; + /** 二次判死线(心跳正常但连续 N 分钟无结果分片上报)阈值,分钟;<=0 关闭本线。 */ + @Value("${aiimage.shop-data-crawl.no-result-upload-timeout-minutes:180}") + private long noResultUploadTimeoutMinutes; + /** * 陈旧任务判死:扫描本模块「RUNNING 且 Python 心跳超时」的任务并终结。 * @@ -175,6 +180,11 @@ public class ShopDataCrawlTaskService { log.warn("[shop-data-crawl] stale task finalization failed taskId={} msg={}", task.getId(), ex.getMessage()); } } + finalizeNoUploadStaleTasks(); + } + + private boolean markStaleTaskFailedIfStillRunning(Long taskId) { + return markTaskFailedIfStillRunning(taskId, "长时间未收到 Python 结果回传,任务已自动失败"); } /** @@ -183,13 +193,13 @@ public class ShopDataCrawlTaskService { * * @return true 表示本次确实由 RUNNING 翻转为 FAILED */ - private boolean markStaleTaskFailedIfStillRunning(Long taskId) { + private boolean markTaskFailedIfStillRunning(Long taskId, String errorMessage) { LocalDateTime now = LocalDateTime.now(); int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() .eq(FileTaskEntity::getId, taskId) .eq(FileTaskEntity::getStatus, "RUNNING") .set(FileTaskEntity::getStatus, "FAILED") - .set(FileTaskEntity::getErrorMessage, "长时间未收到 Python 结果回传,任务已自动失败") + .set(FileTaskEntity::getErrorMessage, errorMessage) .set(FileTaskEntity::getUpdatedAt, now) .set(FileTaskEntity::getFinishedAt, now)); if (updated <= 0) { @@ -202,6 +212,78 @@ public class ShopDataCrawlTaskService { return true; } + /** + * 二次判死:Python 心跳正常(Redis heartbeat 新鲜)但连续 N 分钟无结果分片上报。 + * + *

既有心跳线候选条件是心跳/updated_at 陈旧,而 Python 的 HTTP 心跳线程在任务主线程 + * 卡死时照常发送——心跳永远新鲜,任务永远命不中(与生产 28131 同型缺口)。 + * 本线候选取「心跳新鲜 + 创建超过 N 分钟」,判据用 biz_task_scope_state.last_chunk_at + * (仅分片上传时刷新);从未上报(无 scope 行或全 NULL)的任务跳过,避免误杀首批较慢的正常任务。 + */ + private void finalizeNoUploadStaleTasks() { + long minutes = noResultUploadTimeoutMinutes; + if (minutes <= 0) { + return; + } + long heartbeatFreshMillis = Duration.ofMinutes(Math.max(1L, staleTimeoutMinutes)).toMillis(); + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes); + long nowMillis = System.currentTimeMillis(); + List tasks; + try { + tasks = fileTaskMapper.selectList(new LambdaQueryWrapper() + .eq(FileTaskEntity::getModuleType, MODULE_TYPE) + .eq(FileTaskEntity::getStatus, "RUNNING") + .ge(FileTaskEntity::getUpdatedAt, LocalDateTime.now().minusMinutes(Math.max(1L, staleTimeoutMinutes))) + .lt(FileTaskEntity::getCreatedAt, cutoff) + .last("limit 200")); + } catch (Exception ex) { + log.warn("[shop-data-crawl] no-upload stale task scan failed msg={}", ex.getMessage()); + return; + } + if (tasks.isEmpty()) { + return; + } + Map heartbeats = taskCacheService.getTaskHeartbeatMillisBatch( + tasks.stream().map(FileTaskEntity::getId).toList()); + Map lastResultAtByTaskId = new LinkedHashMap<>(); + for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds( + tasks.stream().map(FileTaskEntity::getId).toList())) { + if (dto.taskId() != null && dto.lastChunkAt() != null) { + lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt()); + } + } + for (FileTaskEntity task : tasks) { + long heartbeat = heartbeats.getOrDefault(task.getId(), 0L); + boolean heartbeatFresh = heartbeat > 0 && nowMillis - heartbeat < heartbeatFreshMillis; + if (!heartbeatFresh) { + // 心跳已陈旧:归既有心跳线处理(updated_at 同源陈旧,由既有候选捞走) + continue; + } + LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId()); + if (lastResultAt == null || lastResultAt.isAfter(cutoff)) { + continue; + } + TaskDistributedLockService.LockHandle taskLock = acquireTaskLock(task.getId(), 0L); + if (taskLock == null) { + log.info("[shop-data-crawl] no-upload stale task skipped because task lock is busy taskId={}", task.getId()); + continue; + } + try (taskLock) { + if (taskFileJobService.countUnfinishedAssembleJobs(task.getId(), MODULE_TYPE) > 0L) continue; + if (!tryFinalizeTask(task.getId(), true, true)) { + markTaskFailedIfStillRunning(task.getId(), + "连续 " + minutes + " 分钟无结果回传,疑似客户端任务卡死,任务已自动失败(最后结果上报 " + lastResultAt + ")"); + } + log.warn("[shop-data-crawl] no-upload stale task finalized taskId={} lastResultAt={} timeoutMinutes={}", + task.getId(), lastResultAt, minutes); + } catch (TaskOwnerMismatchException ignored) { + // The owner may change between the scan and finalization. + } catch (Exception ex) { + log.warn("[shop-data-crawl] no-upload stale task finalization failed taskId={} msg={}", task.getId(), ex.getMessage()); + } + } + } + private FileTaskEntity loadTaskForExecution(Long taskId) { Map cachedTasks = taskCacheService.getTaskCacheBatch(List.of(taskId)); FileTaskEntity cached = cachedTasks.get(taskId); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 6a0312ca..4d5e5c54 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -63,6 +63,7 @@ import com.nanri.aiimage.modules.task.mapper.FileResultMapper; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; import com.nanri.aiimage.modules.task.model.entity.FileResultEntity; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; @@ -98,6 +99,7 @@ import java.util.ArrayList; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.LinkedHashSet; +import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; @@ -859,6 +861,7 @@ public class SimilarAsinTaskService implements SimilarAsinPipelineHost { finalizeStaleTask(task.getId(), "Python interrupted before uploading final similar ASIN result"); } } + finalizeNoUploadStaleTasks(); } public void debugFinalizeStaleTask(Long taskId) { @@ -906,6 +909,57 @@ public class SimilarAsinTaskService implements SimilarAsinPipelineHost { return updatedMillis <= thresholdMillis; } + /** + * 二次判死:Python 心跳正常(Redis heartbeat 新鲜)但连续 N 分钟无结果分片上报。 + * + *

既有判定以 Redis heartbeat 为 stale 主信号(P1-7),但该心跳随 Python 的 HTTP 心跳 + * 每分钟刷新——主线程卡死时心跳线程照发,任务永远不 stale(与生产 28131 同型缺口)。 + * 本线改看 biz_task_scope_state.last_chunk_at(仅在 persistSubmittedChunk 上传分片时刷新), + * 命中后走既有 finalizeStaleTask(封口上传 + LLM 收尾,不粗暴杀任务)。 + * 从未上报(无 scope 行或全 NULL)的任务跳过,避免误杀首批较慢的正常任务。 + */ + private void finalizeNoUploadStaleTasks() { + long minutes = properties.getNoResultUploadTimeoutMinutes(); + if (minutes <= 0) { + return; + } + List tasks = listStaleFinalizeCandidates(); + if (tasks.isEmpty()) { + return; + } + LocalDateTime cutoff = LocalDateTime.now().minusMinutes(minutes); + long heartbeatThresholdMillis = LocalDateTime.now() + .minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())) + .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); + Map lastResultAtByTaskId = new HashMap<>(); + for (TaskScopeLastChunkDto dto : taskScopeStateMapper.selectLastChunkAtByTaskIds( + tasks.stream().map(FileTaskEntity::getId).toList())) { + if (dto.taskId() != null && dto.lastChunkAt() != null) { + lastResultAtByTaskId.put(dto.taskId(), dto.lastChunkAt()); + } + } + for (FileTaskEntity task : tasks) { + if (isHeartbeatStale(task, heartbeatThresholdMillis)) { + // 心跳已 stale:归既有心跳线处理 + continue; + } + LocalDateTime lastResultAt = lastResultAtByTaskId.get(task.getId()); + if (lastResultAt == null || lastResultAt.isAfter(cutoff)) { + continue; + } + TaskDistributedLockService.LockHandle taskLockHandle = ownershipSupport().acquireTaskLock(task.getId(), 0L); + if (taskLockHandle == null) { + continue; + } + try (taskLockHandle) { + log.warn("[similar-asin] 心跳正常但 {} 分钟无结果分片上报,按卡死收尾 taskId={} lastResultAt={}", + minutes, task.getId(), lastResultAt); + finalizeStaleTask(task.getId(), + "Python heartbeat alive but no result chunk uploaded for " + minutes + " minutes"); + } + } + } + private void finalizeStaleTask(Long taskId, String error) { FileTaskEntity task = fileTaskMapper.selectById(taskId); if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !STATUS_RUNNING.equals(task.getStatus())) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java index 3ad6d6f3..261f0a36 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java @@ -3,6 +3,7 @@ package com.nanri.aiimage.modules.task.controller; import com.nanri.aiimage.common.api.ApiResponse; import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest; import com.nanri.aiimage.modules.task.model.vo.TaskHeartbeatVo; +import com.nanri.aiimage.modules.task.service.TaskHeartbeatPositionService; import com.nanri.aiimage.modules.task.service.TaskHeartbeatService; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.Parameter; @@ -22,6 +23,7 @@ import org.springframework.web.bind.annotation.RestController; public class TaskHeartbeatController { private final TaskHeartbeatService taskHeartbeatService; + private final TaskHeartbeatPositionService taskHeartbeatPositionService; @PostMapping("/{taskId}/heartbeat") @Operation( @@ -31,6 +33,8 @@ public class TaskHeartbeatController { @Parameter(description = "任务 ID", required = true, example = "200") @PathVariable Long taskId, @Valid @RequestBody(required = false) TaskHeartbeatRequest request) { + // 先记录「当前处理位置」(仅用于判死文案与排查,内部异常降级,不影响心跳本身) + taskHeartbeatPositionService.record(taskId, request); return ApiResponse.success(taskHeartbeatService.heartbeat(taskId, request)); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskScopeStateMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskScopeStateMapper.java index c4c49612..f287182f 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskScopeStateMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskScopeStateMapper.java @@ -1,9 +1,34 @@ package com.nanri.aiimage.modules.task.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; import org.apache.ibatis.annotations.Mapper; +import org.apache.ibatis.annotations.Param; +import org.apache.ibatis.annotations.Select; + +import java.util.List; @Mapper public interface TaskScopeStateMapper extends BaseMapper { + + /** + * 各任务「最后一次结果分片上报时间」(二次判死线专用)。 + * + *

last_chunk_at 只随结果分片上报刷新(TaskScopePayloadStorageService.saveScopePayload), + * 不随 HTTP 心跳刷新——是区分「客户端进程活着」与「任务有进展」的可靠信号。 + * HAVING 过滤掉全 NULL 的组:从未上报的任务不出现在结果里,调用方据此跳过(不判死)。 + * GROUP BY 走 uk_task_scope 的 task_id 前缀。 + */ + @Select(""" + + """) + List selectLastChunkAtByTaskIds(@Param("taskIds") List taskIds); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java index f8f0d34a..105c1373 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java @@ -38,4 +38,8 @@ public class TaskHeartbeatRequest { @Schema(description = "详情页 ASIN 总数", example = "120") private Integer detailTotalAsins; + + @Schema(description = "客户端归一化后的当前处理位置描述(如「店铺 张三(2/10),国家 德国(1/3)」)," + + "用于判死日志与排查;老客户端不传,服务端不得以位置为空作为判死依据", example = "店铺 魏振峰(2/10)") + private String progressText; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskScopeLastChunkDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskScopeLastChunkDto.java new file mode 100644 index 00000000..c013263d --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskScopeLastChunkDto.java @@ -0,0 +1,12 @@ +package com.nanri.aiimage.modules.task.model.dto; + +import java.time.LocalDateTime; + +/** + * 任务「最后一次结果分片上报时间」(「心跳正常但无结果上报」二次判死线用)。 + * + *

lastChunkAt 取 biz_task_scope_state 的 MAX(last_chunk_at)——该列只随结果分片上报刷新 + * (TaskScopePayloadStorageService.saveScopePayload),不随 HTTP 心跳刷新。 + */ +public record TaskScopeLastChunkDto(Long taskId, LocalDateTime lastChunkAt) { +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatPositionService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatPositionService.java new file mode 100644 index 00000000..58fcc38a --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatPositionService.java @@ -0,0 +1,98 @@ +package com.nanri.aiimage.modules.task.service; + +import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.stereotype.Service; + +import java.time.Duration; +import java.time.Instant; +import java.util.HashMap; +import java.util.Map; + +/** + * 任务心跳的「当前处理位置」存储(Redis Hash,24h TTL)。 + * + *

客户端主线程(浏览器自动化)卡死时心跳线程照发,仅凭心跳无法区分 + * 「进程活着」与「任务有进展」。位置文本用于判死日志 / errorMessage 说明卡在哪, + * 以及后台人工排查。位置缺失只退化文案,不得作为任何判死依据。 + * 老客户端不传 progressText、位置全空时不写;读取失败返回 null。 + */ +@Slf4j +@Service +@RequiredArgsConstructor +public class TaskHeartbeatPositionService { + + private static final String KEY_PREFIX = "task:heartbeat:position:"; + private static final Duration POSITION_TTL = Duration.ofHours(24); + + private final StringRedisTemplate stringRedisTemplate; + + public void record(Long taskId, TaskHeartbeatRequest request) { + if (taskId == null || taskId <= 0 || request == null) { + return; + } + String progressText = request.getProgressText() == null ? "" : request.getProgressText().trim(); + boolean hasPosition = !progressText.isEmpty() + || request.getCurrent() != null + || request.getTotal() != null + || (request.getPhase() != null && !request.getPhase().isBlank()); + if (!hasPosition) { + return; + } + try { + Map values = new HashMap<>(); + values.put("phase", request.getPhase() == null ? "" : request.getPhase()); + values.put("current", request.getCurrent() == null ? "" : String.valueOf(request.getCurrent())); + values.put("total", request.getTotal() == null ? "" : String.valueOf(request.getTotal())); + values.put("progressText", progressText); + values.put("at", String.valueOf(Instant.now().toEpochMilli())); + String key = KEY_PREFIX + taskId; + stringRedisTemplate.opsForHash().putAll(key, values); + stringRedisTemplate.expire(key, POSITION_TTL); + } catch (Exception ex) { + log.warn("[heartbeat-position] 记录位置失败 taskId={} msg={}", taskId, ex.getMessage()); + } + } + + /** 读取「最后处理位置」描述;无位置信息返回 null(调用方自行降级)。 */ + public String describe(Long taskId) { + if (taskId == null || taskId <= 0) { + return null; + } + Map entries; + try { + entries = stringRedisTemplate.opsForHash().entries(KEY_PREFIX + taskId); + } catch (Exception ex) { + log.warn("[heartbeat-position] 读取位置失败 taskId={} msg={}", taskId, ex.getMessage()); + return null; + } + if (entries == null || entries.isEmpty()) { + return null; + } + String text = value(entries, "progressText"); + if (!text.isEmpty()) { + return text; + } + String phase = value(entries, "phase"); + String current = value(entries, "current"); + String total = value(entries, "total"); + StringBuilder sb = new StringBuilder(); + if (!phase.isEmpty()) { + sb.append("阶段=").append(phase); + } + if (!current.isEmpty() || !total.isEmpty()) { + if (sb.length() > 0) { + sb.append(","); + } + sb.append(current.isEmpty() ? "0" : current).append("/").append(total.isEmpty() ? "0" : total); + } + return sb.length() == 0 ? null : sb.toString(); + } + + private String value(Map entries, String key) { + Object raw = entries.get(key); + return raw == null ? "" : String.valueOf(raw).trim(); + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/controller/TutorialAdminController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/controller/TutorialAdminController.java index 0107c34a..04ddd82e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/controller/TutorialAdminController.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/controller/TutorialAdminController.java @@ -46,10 +46,12 @@ public class TutorialAdminController { @Operation(summary = "签发教程包直传预签名", description = "返回浏览器直传 MinIO client 桶的 PUT 预签名地址;直传完成后调用 /tutorial/confirm 校验落库") public ApiResponse> presignTutorial( HttpServletRequest request, - @Parameter(description = "上传的 zip 文件名") @RequestParam(value = "file_name", required = false) String fileName) { + @Parameter(description = "上传的 zip 文件名") @RequestParam(value = "file_name", required = false) String fileName, + @Parameter(description = "教程包版本号(展示与排序用,可空)") @RequestParam(value = "version", required = false) String version) { AdminUserEntity operator = adminAuthSupport.requireAdminOrInternal(request); - log.info("[tutorial] 管理端签发教程包直传预签名 operator={} fileName={}", operator.getUsername(), fileName); - return ApiResponse.success(tutorialPackageService.presignTutorialPackage(fileName)); + log.info("[tutorial] 管理端签发教程包直传预签名 operator={} fileName={} version={}", + operator.getUsername(), fileName, version); + return ApiResponse.success(tutorialPackageService.presignTutorialPackage(fileName, version)); } @PostMapping("/tutorial/confirm") @@ -57,11 +59,12 @@ public class TutorialAdminController { public ApiResponse> confirmTutorial( HttpServletRequest request, @Parameter(description = "预签名返回的对象 key") @RequestParam(value = "object_key", required = false) String objectKey, - @Parameter(description = "上传的 zip 文件名") @RequestParam(value = "file_name", required = false) String fileName) { + @Parameter(description = "上传的 zip 文件名") @RequestParam(value = "file_name", required = false) String fileName, + @Parameter(description = "教程包版本号(展示与排序用,可空)") @RequestParam(value = "version", required = false) String version) { AdminUserEntity operator = adminAuthSupport.requireAdminOrInternal(request); - log.info("[tutorial] 管理端确认教程包直传完成 operator={} objectKey={} fileName={}", - operator.getUsername(), objectKey, fileName); - return ApiResponse.success("上传成功", Map.of("item", tutorialPackageService.confirmTutorialPackage(objectKey, fileName))); + log.info("[tutorial] 管理端确认教程包直传完成 operator={} objectKey={} fileName={} version={}", + operator.getUsername(), objectKey, fileName, version); + return ApiResponse.success("上传成功", Map.of("item", tutorialPackageService.confirmTutorialPackage(objectKey, fileName, version))); } @PostMapping("/tutorial/delete") diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/model/entity/TutorialPackageEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/model/entity/TutorialPackageEntity.java index e3bcda42..5f81b460 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/model/entity/TutorialPackageEntity.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/model/entity/TutorialPackageEntity.java @@ -22,6 +22,9 @@ public class TutorialPackageEntity { /** 原始文件名(展示用) */ private String fileName; + /** 教程包版本号(上传时填写;V126 之前的历史行可为空) */ + private String version; + /** MinIO 对象 key(client 桶,tutorial/ 前缀) */ private String objectKey; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/service/TutorialPackageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/service/TutorialPackageService.java index e5672931..e5dd05fc 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/service/TutorialPackageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/tutorial/service/TutorialPackageService.java @@ -37,6 +37,9 @@ public class TutorialPackageService { /** 教程包与软件版本包同量级,沿用 512MB 上限 */ private static final long MAX_UPLOAD_BYTES = 512L * 1024 * 1024; + /** 版本号长度上限(与 biz_tutorial_package.version VARCHAR(64) 对齐) */ + private static final int MAX_VERSION_LENGTH = 64; + /** 直传预签名有效期:3 分钟足够数百 MB 级压缩包上传 */ private static final int PRESIGN_EXPIRY_SECONDS = 180; @@ -76,15 +79,17 @@ public class TutorialPackageService { if (entity == null) { log.info("[tutorial] 查询最新教程包:无记录,前端将回退固定直链"); result.put("file_name", null); + result.put("version", null); result.put("file_url", null); result.put("file_size", null); result.put("created_at", null); return result; } Map item = toItemMap(entity); - log.info("[tutorial] 查询最新教程包 id={} fileName={} objectKey={}", - entity.getId(), entity.getFileName(), entity.getObjectKey()); + log.info("[tutorial] 查询最新教程包 id={} fileName={} version={} objectKey={}", + entity.getId(), entity.getFileName(), entity.getVersion(), entity.getObjectKey()); result.put("file_name", item.get("file_name")); + result.put("version", item.get("version")); result.put("file_url", item.get("file_url")); result.put("file_size", item.get("file_size")); result.put("created_at", item.get("created_at")); @@ -94,27 +99,32 @@ public class TutorialPackageService { /** * 签发教程包直传预签名:按 时间戳-原文件名 生成新对象 key(每次上传互不覆盖), * 浏览器直传到 MinIO 后再调 {@link #confirmTutorialPackage} 由服务端核对落库。 + *

版本号为上传时填写的展示信息,一并回传给前端在 confirm 阶段落库。 */ - public Map presignTutorialPackage(String fileName) { + public Map presignTutorialPackage(String fileName, String version) { String normalizedName = requireZipFileName(fileName); + String normalizedVersion = normalizeVersion(version); String objectKey = OBJECT_KEY_PREFIX + OBJECT_NAME_FORMATTER.format(LocalDateTime.now()) + "-" + safeFileName(normalizedName); - log.info("[tutorial] 签发教程包直传预签名 fileName={} objectKey={} expirySeconds={}", - normalizedName, objectKey, PRESIGN_EXPIRY_SECONDS); + log.info("[tutorial] 签发教程包直传预签名 fileName={} version={} objectKey={} expirySeconds={}", + normalizedName, normalizedVersion, objectKey, PRESIGN_EXPIRY_SECONDS); String uploadUrl = ossStorageService.presignTutorialUpload(objectKey, PRESIGN_EXPIRY_SECONDS); Map result = new LinkedHashMap<>(); result.put("object_key", objectKey); result.put("upload_url", uploadUrl); result.put("file_url", ossStorageService.getTutorialDownloadUrl(objectKey)); + result.put("version", normalizedVersion); return result; } /** 直传完成后确认落库:校验对象存在且不超限,新 key 不与其它记录冲突,写入记录并返回新行。 */ - public Map confirmTutorialPackage(String objectKey, String fileName) { + public Map confirmTutorialPackage(String objectKey, String fileName, String version) { if (objectKey == null || !objectKey.startsWith(OBJECT_KEY_PREFIX)) { + log.warn("[tutorial] 直传确认失败:非法对象路径 objectKey={}", objectKey); throw new BusinessException("非法的对象路径,请重新上传"); } String normalizedName = requireZipFileName(fileName); + String normalizedVersion = normalizeVersion(version); long size = ossStorageService.tutorialObjectSize(objectKey); if (size < 0) { log.warn("[tutorial] 直传确认失败:对象不存在 objectKey={}", objectKey); @@ -134,11 +144,12 @@ public class TutorialPackageService { TutorialPackageEntity entity = new TutorialPackageEntity(); entity.setFileName(normalizedName); + entity.setVersion(normalizedVersion); entity.setObjectKey(objectKey); entity.setFileSize(size); tutorialPackageMapper.insert(entity); - log.info("[tutorial] 教程包登记成功 id={} fileName={} objectKey={} bytes={}", - entity.getId(), normalizedName, objectKey, size); + log.info("[tutorial] 教程包登记成功 id={} fileName={} version={} objectKey={} bytes={}", + entity.getId(), normalizedName, normalizedVersion, objectKey, size); TutorialPackageEntity saved = tutorialPackageMapper.selectById(entity.getId()); return toItemMap(saved == null ? entity : saved); } @@ -201,6 +212,20 @@ public class TutorialPackageService { return normalized; } + /** + * 版本号归一:去空白后作为展示字段入库,允许为空(老客户端/历史行兼容)。 + *

与软件版本号不同,教程包版本号只是展示与排序用的标签,不参与对象 key, + * 因此超长时截断而不是直接拒绝上传。 + */ + private String normalizeVersion(String version) { + String normalized = version == null ? "" : version.trim(); + if (normalized.length() > MAX_VERSION_LENGTH) { + log.warn("[tutorial] 版本号超长已截断 length={} limit={}", normalized.length(), MAX_VERSION_LENGTH); + return normalized.substring(0, MAX_VERSION_LENGTH); + } + return normalized; + } + /** 文件名 → 安全对象名片段:去掉路径分隔,只保留中文/字母/数字/._-,空则兜底 tutorial.zip。 */ private String safeFileName(String fileName) { String base = fileName.replace('\\', '/'); @@ -216,6 +241,7 @@ public class TutorialPackageService { Map item = new LinkedHashMap<>(); item.put("id", entity.getId()); item.put("file_name", entity.getFileName() == null ? "" : entity.getFileName()); + item.put("version", entity.getVersion() == null ? "" : entity.getVersion()); item.put("object_key", entity.getObjectKey() == null ? "" : entity.getObjectKey()); item.put("file_size", entity.getFileSize() == null ? 0L : entity.getFileSize()); item.put("file_url", entity.getObjectKey() == null || entity.getObjectKey().isBlank() diff --git a/backend-java/src/main/resources/application-local.example.yml b/backend-java/src/main/resources/application-local.example.yml index 973b46de..5fcb8b7f 100644 --- a/backend-java/src/main/resources/application-local.example.yml +++ b/backend-java/src/main/resources/application-local.example.yml @@ -66,7 +66,7 @@ AIIMAGE_APPEARANCE_PATENT_TITLE_MODEL=gemini-3.8-flash AIIMAGE_APPEARANCE_PATENT_APPEARANCE_MODEL=gemini-3.8-flash AIIMAGE_APPEARANCE_PATENT_LLM_MAX_TOKENS=64000 AIIMAGE_APPEARANCE_PATENT_LLM_BATCH_SIZE=10 -AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY=10 +AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY=5 AIIMAGE_APPEARANCE_PATENT_LLM_RETRY_TIMES=3 AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES=30 diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 08e5b1f4..684bacc9 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -186,6 +186,7 @@ aiimage: failed-ttl-hours: ${AIIMAGE_BRAND_PROGRESS_FAILED_TTL_HOURS:2} heartbeat-timeout-minutes: ${AIIMAGE_BRAND_PROGRESS_HEARTBEAT_TIMEOUT_MINUTES:30} stale-check-cron: ${AIIMAGE_BRAND_PROGRESS_STALE_CHECK_CRON:0 */2 * * * *} + no-result-upload-timeout-minutes: ${AIIMAGE_BRAND_PROGRESS_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180} delete-brand-progress: heartbeat-timeout-minutes: ${AIIMAGE_DELETE_BRAND_HEARTBEAT_TIMEOUT_MINUTES:30} stale-check-cron: ${AIIMAGE_DELETE_BRAND_STALE_CHECK_CRON:*/30 * * * * *} @@ -196,6 +197,8 @@ aiimage: patrol-delete-stale-timeout-minutes: ${AIIMAGE_PATROL_DELETE_STALE_TIMEOUT_MINUTES:30} query-asin-stale-timeout-minutes: ${AIIMAGE_QUERY_ASIN_STALE_TIMEOUT_MINUTES:30} withdraw-stale-timeout-minutes: ${AIIMAGE_WITHDRAW_STALE_TIMEOUT_MINUTES:30} + no-result-upload-timeout-minutes: ${AIIMAGE_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180} + no-result-upload-check-enabled: ${AIIMAGE_NO_RESULT_UPLOAD_CHECK_ENABLED:true} module-cleanup: enabled: ${AIIMAGE_MODULE_CLEANUP_ENABLED:true} cron: ${AIIMAGE_MODULE_CLEANUP_CRON:0 0 0 * * *} @@ -249,11 +252,13 @@ aiimage: path: ${AIIMAGE_BRAND_CHECK_PATH:/brand_check} token: ${AIIMAGE_BRAND_CHECK_TOKEN:} default-strategy: ${AIIMAGE_BRAND_CHECK_DEFAULT_STRATEGY:Terms} - retry-times: ${AIIMAGE_BRAND_CHECK_RETRY_TIMES:3} + retry-times: ${AIIMAGE_BRAND_CHECK_RETRY_TIMES:10} retry-interval-millis: ${AIIMAGE_BRAND_CHECK_RETRY_INTERVAL_MILLIS:1000} + retry-max-interval-millis: ${AIIMAGE_BRAND_CHECK_RETRY_MAX_INTERVAL_MILLIS:10000} connect-timeout-millis: ${AIIMAGE_BRAND_CHECK_CONNECT_TIMEOUT_MILLIS:10000} read-timeout-millis: ${AIIMAGE_BRAND_CHECK_READ_TIMEOUT_MILLIS:60000} appearance-patent: + no-result-upload-timeout-minutes: ${AIIMAGE_APPEARANCE_PATENT_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180} llm-host: ${AIIMAGE_APPEARANCE_PATENT_LLM_HOST:https://ai.t8star.org} title-model: ${AIIMAGE_APPEARANCE_PATENT_TITLE_MODEL:gemini-3.8-flash} appearance-model: ${AIIMAGE_APPEARANCE_PATENT_APPEARANCE_MODEL:gemini-3.8-flash} @@ -262,7 +267,10 @@ aiimage: llm-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_LLM_READ_TIMEOUT_MILLIS:180000} llm-first-attempt-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_LLM_FIRST_ATTEMPT_READ_TIMEOUT_MILLIS:60000} llm-batch-size: ${AIIMAGE_APPEARANCE_PATENT_LLM_BATCH_SIZE:10} - llm-row-concurrency: ${AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY:10} + # 批内行并发:每行会串行发 2 个 LLM 请求(标题提取 + 外观检测),且批次串行提交, + # 故该值≈单任务对 LLM 网关的瞬时并发。原 10 与品牌检测的 8 同一量级, + # 多任务并行时会成倍放大、打满用户自己的网关密钥配额(2026-09-14 随品牌一起下调)。 + llm-row-concurrency: ${AIIMAGE_APPEARANCE_PATENT_LLM_ROW_CONCURRENCY:5} max-parse-rows: ${AIIMAGE_APPEARANCE_PATENT_MAX_PARSE_ROWS:50000} llm-retry-times: ${AIIMAGE_APPEARANCE_PATENT_LLM_RETRY_TIMES:3} flush-pending-minutes: ${AIIMAGE_APPEARANCE_PATENT_FLUSH_PENDING_MINUTES:${AIIMAGE_APPEARANCE_PATENT_COZE_FLUSH_PENDING_MINUTES:1}} @@ -274,6 +282,7 @@ aiimage: db-task-touch-interval-millis: ${AIIMAGE_SIMILAR_ASIN_DB_TASK_TOUCH_INTERVAL_MILLIS:120000} db-job-touch-interval-millis: ${AIIMAGE_SIMILAR_ASIN_DB_JOB_TOUCH_INTERVAL_MILLIS:60000} llm-flush-pending-minutes: ${AIIMAGE_SIMILAR_ASIN_LLM_FLUSH_PENDING_MINUTES:1} + no-result-upload-timeout-minutes: ${AIIMAGE_SIMILAR_ASIN_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180} image-download-pool-size: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_POOL_SIZE:2} image-download-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_TIMEOUT_SECONDS:5} image-prefetch-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_IMAGE_PREFETCH_TIMEOUT_SECONDS:1800} @@ -299,6 +308,7 @@ aiimage: llm-image-download-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_LLM_IMAGE_DOWNLOAD_TIMEOUT_SECONDS:10} collect-data: stale-timeout-minutes: ${AIIMAGE_COLLECT_DATA_STALE_TIMEOUT_MINUTES:30} + no-result-upload-timeout-minutes: ${AIIMAGE_COLLECT_DATA_NO_RESULT_UPLOAD_TIMEOUT_MINUTES:180} stale-check-cron: ${AIIMAGE_COLLECT_DATA_STALE_CHECK_CRON:*/30 * * * * *} max-source-file-bytes: ${AIIMAGE_COLLECT_DATA_MAX_SOURCE_FILE_BYTES:0} max-parse-rows: ${AIIMAGE_COLLECT_DATA_MAX_PARSE_ROWS:0} diff --git a/backend-java/src/main/resources/db/V126__tutorial_package_version.sql b/backend-java/src/main/resources/db/V126__tutorial_package_version.sql new file mode 100644 index 00000000..f30ae4c8 --- /dev/null +++ b/backend-java/src/main/resources/db/V126__tutorial_package_version.sql @@ -0,0 +1,22 @@ +-- V126: biz_tutorial_package 增加 version 列(后台「教程管理」展示与上传时填写的版本号) +-- +-- 背景:教程包后台列表原先只有文件名/大小/上传时间,无法区分同名的多轮教程包; +-- 上传时补填版本号(与软件版本管理页同交互),列表新增「版本号」列并支持排序。 +-- +-- 历史行无版本号:列可空,Java 侧读取为 null 时展示为空(历史种子行不受影响)。 +-- 幂等:ADD COLUMN 前先查 information_schema,重复执行安全。 +-- 回滚:ALTER TABLE biz_tutorial_package DROP COLUMN version; + +SET @db_name = DATABASE(); + +SET @col_exists := ( + SELECT COUNT(*) FROM information_schema.COLUMNS + WHERE TABLE_SCHEMA = @db_name AND TABLE_NAME = 'biz_tutorial_package' AND COLUMN_NAME = 'version' +); +SET @sql := IF(@col_exists = 0, + 'ALTER TABLE biz_tutorial_package ADD COLUMN version VARCHAR(64) NULL COMMENT ''教程包版本号(历史行可为空)'' AFTER file_name', + 'SELECT 1' +); +PREPARE stmt FROM @sql; +EXECUTE stmt; +DEALLOCATE PREPARE stmt; diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataNoUploadStaleTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataNoUploadStaleTest.java new file mode 100644 index 00000000..6c3c2db7 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/service/CollectDataNoUploadStaleTest.java @@ -0,0 +1,143 @@ +package com.nanri.aiimage.modules.collectdata.service; + +import com.nanri.aiimage.modules.task.mapper.FileResultMapper; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; + +import java.time.LocalDateTime; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * collect-data 二次判死线(心跳正常但连续 N 分钟无结果分片上报)。 + * + *

既有心跳线候选条件是 updated_at 陈旧(心跳线),而 Python 心跳线程在主线程卡死时照发—— + * 任务永远命不中。本线候选取「心跳新鲜 + 创建超过 N 分钟」,判据用 + * biz_task_scope_state.last_chunk_at(仅分片上传时刷新);从未上报(无 scope 行)跳过。 + */ +@ExtendWith(MockitoExtension.class) +class CollectDataNoUploadStaleTest { + + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private TaskChunkMapper taskChunkMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskDistributedLockService.LockHandle lockHandle; + + @Spy + private com.fasterxml.jackson.databind.ObjectMapper objectMapper = new com.fasterxml.jackson.databind.ObjectMapper(); + + @InjectMocks + private CollectDataService service; + + @BeforeEach + void setUp() { + ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 30L); + ReflectionTestUtils.setField(service, "noResultUploadTimeoutMinutes", 180L); + ReflectionTestUtils.setField(service, "progressThrottleMillis", 0L); + ReflectionTestUtils.setField(service, "progressDirtyWindowMillis", 0L); + } + + private FileTaskEntity runningTask(long id) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(id); + task.setModuleType("COLLECT_DATA"); + task.setStatus("RUNNING"); + task.setCreatedAt(LocalDateTime.now().minusDays(1)); + task.setUpdatedAt(LocalDateTime.now().minusMinutes(1)); + return task; + } + + private void invokeNoUpload() { + ReflectionTestUtils.invokeMethod(service, "finalizeNoUploadStaleTasks", + LocalDateTime.now().minusMinutes(30)); + } + + @Test + void 上报仍新鲜时不判死() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(runningTask(1L))); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(1L, LocalDateTime.now().minusMinutes(5)))); + + invokeNoUpload(); + + verify(taskDistributedLockService, never()).acquire(any(), anyLong(), anyLong()); + verify(fileTaskMapper, never()).updateById(any()); + } + + @Test + void 从未上报时跳过() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(runningTask(2L))); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())).thenReturn(List.of()); + + invokeNoUpload(); + + verify(taskDistributedLockService, never()).acquire(any(), anyLong(), anyLong()); + verify(fileTaskMapper, never()).updateById(any()); + } + + @Test + void 开关为0时不扫描() { + ReflectionTestUtils.setField(service, "noResultUploadTimeoutMinutes", 0L); + + invokeNoUpload(); + + verify(fileTaskMapper, never()).selectList(any()); + } + + @Test + void 任务锁忙时跳过() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(runningTask(3L))); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(3L, LocalDateTime.now().minusHours(4)))); + when(taskDistributedLockService.acquire(any(), anyLong(), anyLong())).thenReturn(null); + + invokeNoUpload(); + + verify(fileTaskMapper, never()).updateById(any()); + } + + @Test + void 心跳正常但上报超时触发判死() { + FileTaskEntity task = runningTask(5L); + when(fileTaskMapper.selectList(any())).thenReturn(List.of(task)); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(5L, LocalDateTime.now().minusHours(4)))); + when(taskDistributedLockService.acquire(any(), anyLong(), anyLong())).thenReturn(lockHandle); + when(fileTaskMapper.selectById(5L)).thenReturn(task); + when(taskFileJobService.countUnfinishedAssembleJobs(5L, "COLLECT_DATA")).thenReturn(0L); + when(taskChunkMapper.selectCount(any())).thenReturn(0L); + when(fileResultMapper.selectOne(any())).thenReturn(null); + + invokeNoUpload(); + + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(FileTaskEntity.class); + verify(fileTaskMapper).updateById(taskCaptor.capture()); + assertEquals("FAILED", taskCaptor.getValue().getStatus()); + assertTrue(taskCaptor.getValue().getErrorMessage().contains("无结果回传"), + taskCaptor.getValue().getErrorMessage()); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskServiceTest.java index 6d1bca84..1e9f66d4 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskServiceTest.java @@ -5,10 +5,15 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; import com.nanri.aiimage.config.DeleteBrandProgressProperties; +import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskCacheService; +import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskService; import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskScopeLastChunkDto; import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskHeartbeatPositionService; import org.apache.ibatis.builder.MapperBuilderAssistant; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -50,6 +55,10 @@ class DeleteBrandStaleTaskServiceTest { @Mock private DeleteBrandProgressProperties deleteBrandProgressProperties; @Mock private TaskDistributedLockService taskDistributedLockService; @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private ProductRiskTaskService productRiskTaskService; + @Mock private ProductRiskTaskCacheService productRiskTaskCacheService; + @Mock private TaskHeartbeatPositionService taskHeartbeatPositionService; @BeforeAll static void initializeTableInfo() { @@ -212,11 +221,184 @@ class DeleteBrandStaleTaskServiceTest { assertTrue(errorMessage.contains("缺失分片"), errorMessage); } + // ---------- 二次判死线:心跳正常但连续 N 分钟无结果分片上报 ---------- + + @Test + void noUploadStaleTaskIsFailedWithCasAndCacheEvicted() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(901L, "PRODUCT_RISK_RESOLVE"))); + noUploadEnabled(); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(901L, LocalDateTime.now().minusHours(4)))); + when(taskFileJobService.countUnfinishedAssembleJobs(901L, "PRODUCT_RISK_RESOLVE")).thenReturn(0L); + noUploadTaskLockAvailable(); + when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + verify(productRiskTaskService).tryFinalizeTask(901L, true); + verify(productRiskTaskCacheService).deleteTaskCache(901L); + @SuppressWarnings({"rawtypes", "unchecked"}) + ArgumentCaptor update = ArgumentCaptor.forClass(LambdaUpdateWrapper.class); + verify(fileTaskMapper).update(isNull(), update.capture()); + List values = update.getValue().getParamNameValuePairs().values().stream() + .map(String::valueOf) + .toList(); + assertTrue(values.stream().anyMatch(v -> v.contains("无结果回传")), values.toString()); + } + + @Test + void noUploadFreshResultIsSkipped() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(902L, "PRODUCT_RISK_RESOLVE"))); + noUploadEnabled(); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(902L, LocalDateTime.now().minusMinutes(5)))); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + verify(taskDistributedLockService, never()).acquire(any(), any(), anyLong()); + verify(fileTaskMapper, never()).update(any(), any()); + } + + @Test + void noUploadNeverUploadedIsSkipped() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(903L, "PRODUCT_RISK_RESOLVE"))); + noUploadEnabled(); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())).thenReturn(List.of()); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + verify(taskDistributedLockService, never()).acquire(any(), any(), anyLong()); + verify(fileTaskMapper, never()).update(any(), any()); + } + + @Test + void noUploadAlreadyFinalizedTaskKeepsCache() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(904L, "PRODUCT_RISK_RESOLVE"))); + noUploadEnabled(); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(904L, LocalDateTime.now().minusHours(4)))); + when(taskFileJobService.countUnfinishedAssembleJobs(904L, "PRODUCT_RISK_RESOLVE")).thenReturn(0L); + noUploadTaskLockAvailable(); + // CAS 未命中:finalize 已把任务终结(状态不再是 RUNNING) + when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(0); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + verify(productRiskTaskCacheService, never()).deleteTaskCache(anyLong()); + } + + @Test + void noUploadBusyTaskLockIsSkipped() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(905L, "PRODUCT_RISK_RESOLVE"))); + noUploadEnabled(); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(905L, LocalDateTime.now().minusHours(4)))); + when(taskFileJobService.countUnfinishedAssembleJobs(905L, "PRODUCT_RISK_RESOLVE")).thenReturn(0L); + when(taskDistributedLockService.acquire(any(), any(), anyLong())).thenReturn(null); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + verify(productRiskTaskService, never()).tryFinalizeTask(anyLong(), anyBoolean()); + verify(fileTaskMapper, never()).update(any(), any()); + } + + @Test + void noUploadPendingAssembleJobsIsSkipped() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(906L, "PRODUCT_RISK_RESOLVE"))); + noUploadEnabled(); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(906L, LocalDateTime.now().minusHours(4)))); + when(taskFileJobService.countUnfinishedAssembleJobs(906L, "PRODUCT_RISK_RESOLVE")).thenReturn(1L); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + verify(productRiskTaskService, never()).tryFinalizeTask(anyLong(), anyBoolean()); + } + + @Test + void noUploadDisabledByConfigSkipsScan() { + when(deleteBrandProgressProperties.isNoResultUploadCheckEnabled()).thenReturn(false); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + verify(fileTaskMapper, never()).selectList(any()); + } + + @Test + void noUploadDeleteBrandDispatchesToRunService() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(907L, "DELETE_BRAND"))); + noUploadEnabled(); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(907L, LocalDateTime.now().minusHours(4)))); + when(taskFileJobService.countUnfinishedAssembleJobs(907L, "DELETE_BRAND")).thenReturn(0L); + noUploadTaskLockAvailable(); + when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + verify(deleteBrandRunService).tryFinalizeTask(907L, true); + verify(deleteBrandTaskCacheService).delete(907L); + } + + @Test + void noUploadFailReasonIncludesLastPosition() { + when(fileTaskMapper.selectList(any())).thenReturn(List.of(noUploadTask(908L, "PRODUCT_RISK_RESOLVE"))); + noUploadEnabled(); + when(taskScopeStateMapper.selectLastChunkAtByTaskIds(any())) + .thenReturn(List.of(new TaskScopeLastChunkDto(908L, LocalDateTime.now().minusHours(4)))); + when(taskFileJobService.countUnfinishedAssembleJobs(908L, "PRODUCT_RISK_RESOLVE")).thenReturn(0L); + noUploadTaskLockAvailable(); + when(taskHeartbeatPositionService.describe(908L)).thenReturn("店铺 魏振峰(2/10),国家 德国(1/3)"); + when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1); + + DeleteBrandStaleTaskService service = service(); + ReflectionTestUtils.invokeMethod(service, "failNoResultUploadTasks"); + + @SuppressWarnings({"rawtypes", "unchecked"}) + ArgumentCaptor update = ArgumentCaptor.forClass(LambdaUpdateWrapper.class); + verify(fileTaskMapper).update(isNull(), update.capture()); + List values = update.getValue().getParamNameValuePairs().values().stream() + .map(String::valueOf) + .toList(); + assertTrue(values.stream().anyMatch(v -> v.contains("最后处理位置:店铺 魏振峰(2/10)")), values.toString()); + } + + private void noUploadEnabled() { + when(deleteBrandProgressProperties.isNoResultUploadCheckEnabled()).thenReturn(true); + when(deleteBrandProgressProperties.getNoResultUploadTimeoutMinutes()).thenReturn(180L); + } + + private void noUploadTaskLockAvailable() { + when(taskDistributedLockService.acquire(any(), any(), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + } + + private static FileTaskEntity noUploadTask(Long id, String moduleType) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(id); + task.setModuleType(moduleType); + task.setStatus("RUNNING"); + task.setCreatedAt(LocalDateTime.now().minusDays(1)); + task.setUpdatedAt(LocalDateTime.now().minusMinutes(1)); + return task; + } + private DeleteBrandStaleTaskService service() { return new DeleteBrandStaleTaskService( fileTaskMapper, deleteBrandTaskCacheService, deleteBrandTaskStorageService, deleteBrandRunService, - null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, - deleteBrandProgressProperties, null, taskDistributedLockService, taskFileJobService, null); + productRiskTaskService, productRiskTaskCacheService, + null, null, null, null, null, null, null, null, null, null, + null, null, null, + deleteBrandProgressProperties, null, taskDistributedLockService, taskFileJobService, null, + taskScopeStateMapper, taskHeartbeatPositionService); } private void lockAvailable() { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/TaskModuleCoverageTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/TaskModuleCoverageTest.java index dbf772d4..ce4068dc 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/TaskModuleCoverageTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/deletebrand/service/TaskModuleCoverageTest.java @@ -81,7 +81,8 @@ class TaskModuleCoverageTest { mock(AppearancePatentTaskService.class), mock(SimilarAsinTaskService.class), null, null, null, null, - mock(ShopDataCrawlTaskService.class)); + mock(ShopDataCrawlTaskService.class), + null, null); } @Test diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatPositionServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatPositionServiceTest.java new file mode 100644 index 00000000..58bfd977 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatPositionServiceTest.java @@ -0,0 +1,114 @@ +package com.nanri.aiimage.modules.task.service; + +import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.data.redis.core.HashOperations; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.time.Duration; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * 心跳「当前处理位置」存储:写读往返、空位置不写、Redis 异常只降级不抛出。 + * 位置仅用于判死文案与排查,任何异常都不得影响心跳本身。 + */ +@ExtendWith(MockitoExtension.class) +class TaskHeartbeatPositionServiceTest { + + @Mock private StringRedisTemplate stringRedisTemplate; + @Mock private HashOperations hashOperations; + + private TaskHeartbeatPositionService service() { + return new TaskHeartbeatPositionService(stringRedisTemplate); + } + + @Test + void recordWritesHashWithProgressText() { + when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations); + TaskHeartbeatRequest request = new TaskHeartbeatRequest(); + request.setPhase("crawling"); + request.setCurrent(2); + request.setTotal(10); + request.setProgressText("店铺 魏振峰(2/10)"); + + service().record(123L, request); + + @SuppressWarnings("unchecked") + ArgumentCaptor> values = ArgumentCaptor.forClass(Map.class); + verify(hashOperations).putAll(eq("task:heartbeat:position:123"), values.capture()); + assertEquals("店铺 魏振峰(2/10)", values.getValue().get("progressText")); + assertEquals("crawling", values.getValue().get("phase")); + assertEquals("2", values.getValue().get("current")); + assertEquals("10", values.getValue().get("total")); + verify(stringRedisTemplate).expire(eq("task:heartbeat:position:123"), any(Duration.class)); + } + + @Test + void recordSkipsWhenNoPosition() { + TaskHeartbeatRequest request = new TaskHeartbeatRequest(); + + service().record(123L, request); + + verify(stringRedisTemplate, never()).opsForHash(); + } + + @Test + void recordSwallowsRedisError() { + when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations); + doThrow(new RuntimeException("redis down")).when(hashOperations).putAll(any(), any()); + TaskHeartbeatRequest request = new TaskHeartbeatRequest(); + request.setProgressText("店铺 A(1/2)"); + + service().record(123L, request); + + verify(stringRedisTemplate, never()).expire(any(), any(Duration.class)); + } + + @Test + void describePrefersProgressText() { + when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations); + when(hashOperations.entries("task:heartbeat:position:123")) + .thenReturn(Map.of("progressText", "店铺 魏振峰(2/10)", "current", "2", "total", "10")); + + assertEquals("店铺 魏振峰(2/10)", service().describe(123L)); + } + + @Test + void describeBuildsFromPhaseAndCountersWhenNoText() { + when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations); + when(hashOperations.entries("task:heartbeat:position:123")) + .thenReturn(Map.of("phase", "crawling", "current", "2", "total", "10")); + + assertEquals("阶段=crawling,2/10", service().describe(123L)); + } + + @Test + void describeReturnsNullWhenMissing() { + when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations); + when(hashOperations.entries("task:heartbeat:position:123")).thenReturn(Map.of()); + + assertNull(service().describe(123L)); + } + + @Test + void describeReturnsNullOnRedisError() { + when(stringRedisTemplate.opsForHash()).thenReturn(hashOperations); + when(hashOperations.entries("task:heartbeat:position:123")) + .thenThrow(new RuntimeException("redis down")); + + assertNull(service().describe(123L)); + } +}