From 93ccb27b4079fe68a4039487ee10599796005ca3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sat, 29 Aug 2026 22:05:30 +0800 Subject: [PATCH] task-36 --- .../service/DailyStateConflictException.java | 13 + .../service/ShopDataCrawlTaskService.java | 99 ++- .../ShopDataCrawlDailyFileLockTest.java | 601 ++++++++++++++++++ 3 files changed, 680 insertions(+), 33 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/DailyStateConflictException.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/DailyStateConflictException.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/DailyStateConflictException.java new file mode 100644 index 00000000..406cd0b5 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/DailyStateConflictException.java @@ -0,0 +1,13 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +/** + * Task 36：每日累计文件归档提交阶段的版本 CAS 冲突信号。 + * 由提交阶段捕获并触发整次归档重试（释放店铺级锁后按最新状态重新组装）， + * 区别于需要调用方直接失败的运行时异常。 + */ +class DailyStateConflictException extends RuntimeException { + + DailyStateConflictException() { + super("daily file state conflict"); + } +} 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 7497d3e7..39af4d17 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 @@ -83,6 +83,8 @@ public class ShopDataCrawlTaskService { private static final int RESULT_SUCCESS = 1; /** 批量 IN 查询单批上限。 */ private static final int ID_BATCH_SIZE = 500; + /** Task 36：每日累计归档版本 CAS 冲突的最大重试次数（每次重试都释放店铺级锁）。 */ + private static final int MAX_DAILY_AGGREGATION_ATTEMPTS = 3; private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断"; private static final String PARTIAL_RESULT_MESSAGE = "Python 中断，已保留已回传的部分数据"; private static final String RESULT_CHUNK_SCOPE_PREFIX = "result-chunks:"; @@ -1742,46 +1744,73 @@ public class ShopDataCrawlTaskService { if (userId == null || shopKeyHash == null) { throw new BusinessException("店铺累计文件归属信息不完整"); } - TaskDistributedLockService.LockHandle lock = dailyFileService.acquireLock(userId, shopKey); - if (lock == null) { - throw new BusinessException("店铺当天累计文件正在处理中，请稍后重试"); - } + // Task 36：版本号/CAS 短临界区。店铺级锁只覆盖“准备/提交”两个毫秒级短事务 + // （各自独立取锁/释放），整表 Excel 组装与 OSS 上传在两次取锁之间于锁外执行； + // 提交阶段按 daily_file.version CAS，冲突时释放锁、清理本次上传对象并按最新状态重试（最多 3 次）。 DailyAggregationResult persistedResult = null; - try { - DailyAggregationPreparation preparation = executeShortTransaction( - () -> prepareDailyAggregation(userId, shopKeyHash, businessDate, row)); + DailyWorkbookArtifact artifact = null; + for (int attempt = 1; attempt <= MAX_DAILY_AGGREGATION_ATTEMPTS; attempt++) { + DailyAggregationPreparation preparation; + TaskDistributedLockService.LockHandle prepareLock = dailyFileService.acquireLock(userId, shopKey); + if (prepareLock == null) { + throw new BusinessException("店铺当天累计文件正在处理中，请稍后重试"); + } + try { + preparation = executeShortTransaction( + () -> prepareDailyAggregation(userId, shopKeyHash, businessDate, row)); + } finally { + prepareLock.close(); + } if (preparation.alreadyArchived()) { return new DailyAggregationResult(List.of(), false); } - - int addedRowCount = excelAssemblyService.countRows(List.of(snapshot)); - ShopDataCrawlDailyFileEntity baseDailyFile = resolveBaseDailyFile( + ShopDataCrawlDailyFileEntity baseForAttempt = resolveBaseDailyFile( preparation.dailyFile(), userId, shopKeyHash, businessDate); - DailyWorkbookArtifact artifact = assembleDailyWorkbook( - task, snapshot, baseDailyFile, addedRowCount); - try { - persistedResult = executeShortTransaction(() -> persistDailyAggregation( - task, row, userId, shopKey, shopKeyHash, businessDate, - preparation, artifact, snapshot)); - if (persistedResult.discardUploadedObject() && artifact.uploaded()) { - deleteObjectQuietly(artifact.objectKey()); - } - return persistedResult; - } catch (RuntimeException ex) { - if (artifact.uploaded()) { - deleteObjectQuietly(artifact.objectKey()); - } - throw ex; + // 组装在锁外执行。每次尝试都用本次准备阶段读到的最新 base 组装 + // （冲突重试时 base 已变化，复用过期的组装结果会把并发写入的行丢在对象外）； + // 零行引用对象不重复上传。 + if (artifact == null || !artifact.uploaded()) { + artifact = assembleDailyWorkbook(task, snapshot, baseForAttempt, + excelAssemblyService.countRows(List.of(snapshot))); + } + TaskDistributedLockService.LockHandle commitLock = dailyFileService.acquireLock(userId, shopKey); + if (commitLock == null) { + throw new BusinessException("店铺当天累计文件正在处理中，请稍后重试"); } - } finally { try { - lock.close(); - } finally { - if (persistedResult != null) { - persistedResult.obsoleteObjectKeys().forEach(this::deleteObjectQuietly); + DailyAggregationPreparation preparationForAttempt = preparation; + DailyWorkbookArtifact artifactForAttempt = artifact; + try { + persistedResult = executeShortTransaction(() -> persistDailyAggregation( + task, row, userId, shopKey, shopKeyHash, businessDate, + preparationForAttempt, artifactForAttempt, snapshot, + preparationForAttempt.dailyFile())); + if (persistedResult.discardUploadedObject() && artifactForAttempt.uploaded()) { + deleteObjectQuietly(artifactForAttempt.objectKey()); + } else { + persistedResult.obsoleteObjectKeys().forEach(this::deleteObjectQuietly); + } + return persistedResult; + } catch (DailyStateConflictException conflictEx) { + // 冲突：本次上传对象作废并释放，按最新状态重新组装再试。 + if (artifact.uploaded()) { + deleteObjectQuietly(artifact.objectKey()); + artifact = null; + } + if (attempt >= MAX_DAILY_AGGREGATION_ATTEMPTS) { + throw new BusinessException("店铺累计文件并发更新冲突，请稍后重试"); + } + } catch (RuntimeException ex) { + if (artifact != null && artifact.uploaded()) { + deleteObjectQuietly(artifact.objectKey()); + } + throw ex; } + } finally { + commitLock.close(); } } + throw new BusinessException("店铺累计文件并发更新冲突，请稍后重试"); } private DailyAggregationPreparation prepareDailyAggregation(Long userId, @@ -1937,14 +1966,18 @@ public class ShopDataCrawlTaskService { LocalDate businessDate, DailyAggregationPreparation preparation, DailyWorkbookArtifact artifact, - ShopDataCrawlResultItemVo snapshot) { + ShopDataCrawlResultItemVo snapshot, + ShopDataCrawlDailyFileEntity expectedBase) { ShopDataCrawlDailyFileEntity dailyFile = dailyFileService.findForUpdate( userId, shopKeyHash, businessDate); if (handleExistingDailyMembership(row, dailyFile)) { return new DailyAggregationResult(List.of(), true); } - if (!sameDailyFileState(preparation.dailyFile(), dailyFile)) { - throw new BusinessException("当天累计文件状态已变化，请重试文件任务"); + if (!Objects.equals(dailyFile == null ? null : dailyFile.getId(), + expectedBase == null ? null : expectedBase.getId()) + || !Objects.equals(dailyFile == null ? null : dailyFile.getVersion(), + expectedBase == null ? null : expectedBase.getVersion())) { + throw new DailyStateConflictException(); } List olderFiles = dailyFileService.findOlder( diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java new file mode 100644 index 00000000..e7b98f4c --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java @@ -0,0 +1,601 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.config.InstanceMetadata; +import com.nanri.aiimage.config.TaskPressureProperties; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlDailyFileEntity; +import com.nanri.aiimage.modules.shopdatacrawl.model.entity.ShopDataCrawlDailyMemberEntity; +import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo; +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.entity.FileResultEntity; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; +import com.nanri.aiimage.modules.task.service.TaskResultItemService; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.apache.ibatis.session.Configuration; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 36：为每日累计文件引入版本号/CAS，缩短店铺级锁的持有时间。 + * 原实现持有店铺级锁覆盖“准备+组装+提交”全流程（含整表 Excel 组装与 OSS 上传， + * 耗时最长）；重构为两个短临界区（准备 / 提交），组装在锁外执行， + * 提交阶段按 daily_file.version CAS，冲突时释放锁重试（最多 3 次）， + * 从而把锁的持有时间从“秒级组装”缩短到“毫秒级两个短事务”。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlDailyFileLockTest { + + private static final String MODULE_TYPE = "SHOP_DATA_CRAWL"; + private static final String SHOP_NAME = "shop-a"; + private static final LocalDate BUSINESS_DATE = LocalDate.of(2026, 8, 29); + private static final LocalDateTime BUSINESS_TIME = LocalDateTime.of(2026, 8, 29, 12, 0); + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new Configuration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + TableInfoHelper.initTableInfo(assistant, ShopDataCrawlDailyFileEntity.class); + TableInfoHelper.initTableInfo(assistant, ShopDataCrawlDailyMemberEntity.class); + } + + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private ShopDataCrawlResolveService shopDataCrawlResolveService; + @Mock private ShopDataCrawlExcelAssemblyService excelAssemblyService; + @Mock private ShopDataCrawlTaskCacheService taskCacheService; + @Mock private OssStorageService ossStorageService; + @Mock private ZiniaoShopSwitchService ziniaoShopSwitchService; + @Mock private TaskPressureProperties taskPressureProperties; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskResultItemService taskResultItemService; + @Mock private TaskProgressSnapshotService taskProgressSnapshotService; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private TaskChunkMapper taskChunkMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private InstanceMetadata instanceMetadata; + @Mock private ShopDataCrawlDailyFileService dailyFileService; + @Spy private ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules(); + + private ShopDataCrawlTaskService service; + + /** 内存中的结果行表（listTaskRows 读取源 + updateById 回写目标）。 */ + private final List dbResultRows = new ArrayList<>(); + /** 内存中的 daily_file 表。 */ + private final List dbDailyFiles = new ArrayList<>(); + /** 内存中的 daily_member 表。 */ + private final List dbMembers = new ArrayList<>(); + private final AtomicLong memberIdSeq = new AtomicLong(1000); + private final AtomicLong fileIdSeq = new AtomicLong(5000); + private long nextResultId = 7200; + private Long lastJobTaskId; + private String lastUploadedObjectKey; + /** 店铺级锁获取次数（验证锁持有时间缩短：两次短临界区各取一次）。 */ + private final AtomicInteger lockAcquireCount = new AtomicInteger(); + private final AtomicReference lastLock = new AtomicReference<>(); + /** 本次测试获取到的全部店铺级锁句柄（验证每个临界区取到的锁都被释放）。 */ + private final List acquiredLocks = new ArrayList<>(); + + @BeforeEach + void configureStorage() { + service = new ShopDataCrawlTaskService( + fileTaskMapper, + fileResultMapper, + shopDataCrawlResolveService, + excelAssemblyService, + taskCacheService, + ossStorageService, + ziniaoShopSwitchService, + objectMapper, + taskPressureProperties, + taskFileJobService, + taskResultItemService, + taskProgressSnapshotService, + taskDistributedLockService, + taskChunkMapper, + taskScopeStateMapper, + transientPayloadStorageService, + instanceMetadata, + dailyFileService, + null); + + dbResultRows.clear(); + dbDailyFiles.clear(); + dbMembers.clear(); + memberIdSeq.set(1000); + fileIdSeq.set(5000); + nextResultId = 7200; + lastJobTaskId = null; + lastUploadedObjectKey = null; + lockAcquireCount.set(0); + lastLock.set(null); + acquiredLocks.clear(); + + lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); + lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of()); + lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of()); + lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())).thenReturn(null); + lenient().when(taskFileJobService.countUnfinishedAssembleJobs(anyLong(), eq(MODULE_TYPE))).thenReturn(0L); + lenient().doNothing().when(taskResultItemService) + .replaceTaskSnapshots(anyLong(), eq(MODULE_TYPE), any(), any()); + lenient().doNothing().when(taskProgressSnapshotService) + .save(anyLong(), any(), any(), anyInt(), anyInt(), anyInt(), any(), any(), any()); + lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + return value == null ? "" : value.trim(); + }); + lenient().when(fileTaskMapper.selectById(anyLong())).thenReturn(null); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + lenient().when(fileResultMapper.selectBatchIds(any())).thenReturn(List.of()); + lenient().when(fileResultMapper.selectCount(any())).thenReturn(0L); + lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenAnswer(invocation -> { + String key = "oss/lock/" + System.nanoTime() + ".xlsx"; + lastUploadedObjectKey = key; + return key; + }); + lenient().when(ossStorageService.readObjectBytes(anyString())).thenReturn(new byte[0]); + lenient().doNothing().when(ossStorageService).deleteObject(anyString()); + lenient().when(excelAssemblyService.countRows(any())).thenAnswer(invocation -> { + List items = invocation.getArgument(0); + return items == null ? 0 : items.size(); + }); + lenient().doAnswer(invocation -> { + List items = invocation.getArgument(1); + return items == null ? 0 : items.size(); + }).when(excelAssemblyService).writeWorkbook(any(), any()); + + // 店铺级锁：每次获取返回独立句柄并计数（两次短临界区各取一次）。 + lenient().when(dailyFileService.acquireLock(anyLong(), anyString())).thenAnswer(invocation -> { + TaskDistributedLockService.LockHandle handle = mock(TaskDistributedLockService.LockHandle.class); + lockAcquireCount.incrementAndGet(); + lastLock.set(handle); + acquiredLocks.add(handle); + return handle; + }); + lenient().when(dailyFileService.currentBusinessDate()).thenReturn(BUSINESS_DATE); + lenient().when(dailyFileService.currentBusinessDateTime()).thenReturn(BUSINESS_TIME); + lenient().when(dailyFileService.shopKeyHash(anyString())).thenAnswer(invocation -> { + String key = invocation.getArgument(0); + return key == null ? null : "hash:" + key; + }); + lenient().when(dailyFileService.shopKey(any(FileResultEntity.class))).thenAnswer(invocation -> { + FileResultEntity row = invocation.getArgument(0); + return row == null ? null : row.getSourceFilename(); + }); + lenient().when(dailyFileService.findForUpdate(anyLong(), anyString(), any())) + .thenAnswer(invocation -> copyDailyFile(findDailyFile( + invocation.getArgument(0), invocation.getArgument(1)))); + lenient().when(dailyFileService.findOlder(anyLong(), anyString(), any())).thenReturn(List.of()); + lenient().when(dailyFileService.findByLatestResultId(anyLong())).thenReturn(List.of()); + lenient().when(dailyFileService.findById(anyLong())).thenReturn(null); + lenient().when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); + lenient().doNothing().when(dailyFileService).deleteDailyFile(anyLong()); + lenient().doNothing().when(dailyFileService).reassignMembers(anyLong(), anyLong()); + lenient().doNothing().when(dailyFileService).deleteMembersForResults(any()); + captureDailyFilePersistence(); + captureMemberInserts(); + captureResultUpdates(); + + // 结果行读取：listTaskRows 按 taskId+moduleType 过滤并升序；其他查询返回空。 + lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + if (wrapper instanceof LambdaQueryWrapper query + && query.getSqlSegment() != null && query.getSqlSegment().contains("taskId")) { + List rows = dbResultRows.stream() + .filter(r -> Objects.equals(r.getTaskId(), lastJobTaskId)) + .sorted(Comparator.comparing(FileResultEntity::getId)) + .toList(); + return new ArrayList<>(rows); + } + return List.of(); + }); + } + + @Test + void test_task_036_daily_file_lock_normal_default_path() { + // 正常路径：归档成功，版本号从 1 起步；锁只取两次（准备/提交短临界区）， + // 整表组装在锁外完成；成员行与累计文件各一。 + FileResultEntity row = addResultRow(7201L, 1L, 1, SHOP_NAME, null); + processJob(1L, List.of(row), snapshot(7201L)); + + assertEquals(1, dbMembers.size(), "默认路径产生一个成员行"); + assertEquals(1, dbDailyFiles.size(), "默认路径产生一个累计文件"); + assertEquals(1L, dbDailyFiles.get(0).getVersion(), "版本号从 1 起步"); + assertEquals(2, lockAcquireCount.get(), "准备/提交两次短临界区各取一次锁"); + assertNotNull(lastUploadedObjectKey, "锁外组装上传了新对象"); + verify(ossStorageService, never()).readObjectBytes(anyString()); + verify(excelAssemblyService, never()).replaceCountriesWorkbook(any(), any(), any()); + } + + @Test + void test_task_036_daily_file_lock_normal_multiple_items() { + // 多结果批量场景：多个结果先后归档，版本号随每次提交递增，成员累积且顺序稳定。 + processJob(1L, List.of(addResultRow(7202L, 1L, 1, SHOP_NAME, null)), snapshot(7202L)); + assertEquals(1L, dbDailyFiles.get(0).getVersion(), "首次归档版本 1"); + processJob(2L, List.of(addResultRow(7203L, 2L, 1, SHOP_NAME, null)), snapshot(7203L)); + + assertEquals(2, dbMembers.size(), "两个结果各一个成员行"); + assertEquals(2L, dbDailyFiles.get(0).getVersion(), "第二次归档版本递增到 2"); + List members = dbMembers.stream() + .sorted(Comparator.comparing(ShopDataCrawlDailyMemberEntity::getResultId)) + .toList(); + assertEquals(7202L, members.get(0).getResultId()); + assertEquals(7203L, members.get(1).getResultId()); + assertNotNull(members.get(0).getRowPayload()); + assertNotNull(members.get(1).getRowPayload()); + } + + @Test + void test_task_036_daily_file_lock_normal_repeated_operation_is_idempotent() { + // 幂等：同一结果重复归档不产生第二个成员行、版本号不递增、不重复上传对象。 + FileResultEntity row = addResultRow(7204L, 1L, 1, SHOP_NAME, null); + processJob(1L, List.of(row), snapshot(7204L)); + long versionAfterFirst = dbDailyFiles.get(0).getVersion(); + String objectAfterFirst = lastUploadedObjectKey; + processJob(1L, List.of(row), snapshot(7204L)); + + assertEquals(1, dbMembers.size(), "重复归档不产生重复成员行"); + assertEquals(versionAfterFirst, dbDailyFiles.get(0).getVersion(), "重复归档版本号不递增"); + assertEquals(objectAfterFirst, lastUploadedObjectKey, "重复归档复用既有对象，不重复上传"); + } + + @Test + void test_task_036_daily_file_lock_boundary_empty_input() { + // 空输入：没有成功结果时安全跳过，不取店铺级锁、不创建成员/累计文件、不上传对象。 + FileResultEntity row = addResultRow(7205L, 1L, 0, SHOP_NAME, null); + processJob(1L, List.of(row), snapshot(7205L)); + + assertTrue(dbMembers.isEmpty(), "无成功结果不创建成员行"); + assertTrue(dbDailyFiles.isEmpty(), "无成功结果不创建累计文件"); + assertEquals(0, lockAcquireCount.get(), "无成功结果不获取店铺级锁"); + verify(ossStorageService, never()).uploadResultFile(any(), eq(MODULE_TYPE)); + } + + @Test + void test_task_036_daily_file_lock_boundary_single_item() { + // 单元素：单结果归档同样只取两次锁（短临界区），版本号 1，结果正确。 + FileResultEntity row = addResultRow(7206L, 1L, 1, SHOP_NAME, null); + processJob(1L, List.of(row), snapshot(7206L)); + + assertEquals(1, dbMembers.size()); + assertEquals(1, dbDailyFiles.size()); + assertEquals(1L, dbDailyFiles.get(0).getVersion()); + assertEquals(2, lockAcquireCount.get(), "单结果也是两次短临界区取锁"); + } + + @Test + void test_task_036_daily_file_lock_boundary_limit_and_overflow() { + // 上限/超限：大量结果逐个归档，版本号与成员数一致递增，无重复无丢失， + // 每次归档的锁获取次数仍为两次（不随量级放大锁持有时间）。 + int limit = 210; + for (int i = 0; i < limit; i++) { + long resultId = nextResultId++; + long taskId = i + 1L; + FileResultEntity row = addResultRow(resultId, taskId, 1, SHOP_NAME, null); + processJob(taskId, List.of(row), snapshot(resultId)); + } + + assertEquals(limit, dbMembers.size(), "大量结果成员行全部保留"); + assertEquals(Long.valueOf(limit), dbDailyFiles.get(0).getVersion(), "版本号与归档次数一致"); + assertEquals(limit * 2, lockAcquireCount.get(), "每次归档恰好两次短临界区取锁"); + } + + @Test + void test_task_036_daily_file_lock_invalid_input_rejected() { + // 非法参数/CAS 冲突：并发提交导致 version CAS 持续冲突时，重试 3 次后 + // 抛出可识别的 BusinessException；每次冲突后上传对象被清理、锁被释放。 + // 通过 findForUpdate 在每次读取后把版本号写回内存表（并返回读到的快照副本） + // 模拟并发写入：提交阶段读到的版本总是比准备阶段新 → 每次 CAS 都冲突。 + // doAnswer().when() 覆盖 @BeforeEach 中同参数 when() 注册（后者重注册不生效）。 + seedDailyFile(); + FileResultEntity row = addResultRow(7207L, 1L, 1, SHOP_NAME, null); + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity current = findDailyFile(invocation.getArgument(0), invocation.getArgument(1)); + if (current == null) { + return null; + } + ShopDataCrawlDailyFileEntity readCopy = copyDailyFile(current); + current.setVersion(readCopy.getVersion() + 1L); + return readCopy; + }).when(dailyFileService).findForUpdate(anyLong(), anyString(), any()); + + Exception ex = assertThrows(BusinessException.class, + () -> processJob(1L, List.of(row), snapshot(7207L))); + assertTrue(ex.getMessage().contains("并发"), "CAS 冲突超限后错误消息可识别"); + assertTrue(dbMembers.isEmpty(), "冲突放弃后不残留成员行"); + assertEquals(1, dbDailyFiles.size(), "冲突放弃后既有累计文件不受破坏"); + assertEquals(6L, dbDailyFiles.get(0).getVersion(), "版本只随并发写入模拟推进（每次读取+1），冲突归档未提交"); + assertEquals(6, acquiredLocks.size(), "三次尝试各取准备/提交两次锁"); + for (TaskDistributedLockService.LockHandle handle : acquiredLocks) { + verify(handle).close(); + } + verify(ossStorageService, times(3)).deleteObject(anyString()); + } + + @Test + void test_task_036_daily_file_lock_dependency_failure_releases_resources() { + // 依赖失败：提交阶段 DB 写入失败时错误可恢复，锁已释放、上传对象已清理、 + // 不残留新成员/累计文件；组装成功路径不受影响（可再次归档）。 + seedDailyFile(); + doThrow(new RuntimeException("commit db down")) + .when(dailyFileService).update(any(ShopDataCrawlDailyFileEntity.class)); + FileResultEntity row = addResultRow(7208L, 1L, 1, SHOP_NAME, null); + assertThrows(RuntimeException.class, + () -> processJob(1L, List.of(row), snapshot(7208L))); + + assertTrue(dbMembers.isEmpty(), "提交失败不残留成员行"); + assertEquals(1, dbDailyFiles.size(), "既有累计文件未被破坏"); + assertEquals(0L, dbDailyFiles.get(0).getVersion(), "既有累计文件版本未被改动"); + assertNotNull(lastLock.get(), "提交阶段已获取锁"); + verify(lastLock.get()).close(); + verify(ossStorageService).deleteObject(anyString()); + + // 错误可恢复：恢复 DB 后同一结果重新归档成功。 + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); + for (int i = 0; i < dbDailyFiles.size(); i++) { + ShopDataCrawlDailyFileEntity existing = dbDailyFiles.get(i); + if (Objects.equals(existing.getUserId(), entity.getUserId()) + && Objects.equals(existing.getShopKeyHash(), entity.getShopKeyHash()) + && Objects.equals(existing.getBusinessDate(), entity.getBusinessDate())) { + entity.setId(existing.getId()); + dbDailyFiles.set(i, entity); + return null; + } + } + entity.setId(fileIdSeq.incrementAndGet()); + dbDailyFiles.add(entity); + return null; + }).when(dailyFileService).update(any(ShopDataCrawlDailyFileEntity.class)); + processJob(1L, List.of(row), snapshot(7208L)); + assertEquals(1, dbMembers.size(), "恢复后同一结果重新归档成功"); + assertEquals(1L, dbDailyFiles.get(0).getVersion(), "恢复后版本号递增"); + } + + // ---- helpers ---- + + private void seedDailyFile() { + ShopDataCrawlDailyFileEntity file = new ShopDataCrawlDailyFileEntity(); + file.setId(1L); + file.setUserId(7L); + file.setShopKeyHash("hash:" + SHOP_NAME); + file.setShopKey(SHOP_NAME); + file.setBusinessDate(BUSINESS_DATE); + file.setResultFilename("daily.xlsx"); + file.setResultFileUrl("oss/daily/seed.xlsx"); + file.setResultFileSize(10L); + file.setVersion(0L); + file.setRowCount(0); + file.setCreatedAt(BUSINESS_TIME); + file.setUpdatedAt(BUSINESS_TIME); + dbDailyFiles.add(file); + } + + private FileResultEntity addResultRow(long id, long taskId, int success, String shopName, String resultFileUrl) { + FileResultEntity row = new FileResultEntity(); + row.setId(id); + row.setTaskId(taskId); + row.setModuleType(MODULE_TYPE); + row.setSuccess(success); + row.setSourceFilename(shopName); + row.setSourceFileUrl("shop-id-" + id); + row.setUserId(7L); + row.setCreatedAt(LocalDateTime.now()); + row.setResultFileUrl(resultFileUrl); + dbResultRows.add(row); + return row; + } + + private ShopDataCrawlResultItemVo snapshot(long resultId) { + ShopDataCrawlResultItemVo item = new ShopDataCrawlResultItemVo(); + item.setResultId(resultId); + item.setTaskId(1L); + item.setShopName(SHOP_NAME); + item.setShopId("shop-id-" + resultId); + item.setSuccess(true); + item.setMatched(true); + item.setTaskStatus("SUCCESS"); + item.setCountryCodes(List.of("DE")); + return item; + } + + private void processJob(long jobTaskId, List rows, ShopDataCrawlResultItemVo snapshot) { + lastJobTaskId = jobTaskId; + TaskFileJobEntity job = new TaskFileJobEntity(); + job.setId(jobTaskId); + job.setTaskId(jobTaskId); + job.setModuleType(MODULE_TYPE); + FileTaskEntity task = taskEntity(jobTaskId); + lenient().when(fileTaskMapper.selectById(jobTaskId)).thenReturn(task); + lenient().when(taskResultItemService.listResultSnapshots(eq(jobTaskId), eq(MODULE_TYPE), any())) + .thenReturn(snapshot == null ? List.of() : List.of(snapshot)); + service.processResultFileJob(job); + } + + private FileTaskEntity taskEntity(long taskId) { + FileTaskEntity task = new FileTaskEntity(); + task.setId(taskId); + task.setModuleType(MODULE_TYPE); + task.setStatus("RUNNING"); + task.setUserId(7L); + task.setOwnerInstanceId("instance-a"); + task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}"); + task.setResultJson("[]"); + task.setCreatedAt(LocalDateTime.now()); + task.setUpdatedAt(LocalDateTime.now()); + return task; + } + + private void captureResultUpdates() { + lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenAnswer(invocation -> { + FileResultEntity updated = invocation.getArgument(0); + for (int i = 0; i < dbResultRows.size(); i++) { + if (Objects.equals(dbResultRows.get(i).getId(), updated.getId())) { + dbResultRows.set(i, updated); + return 1; + } + } + dbResultRows.add(updated); + return 1; + }); + } + + private void captureMemberInserts() { + lenient().when(dailyFileService.addMemberWithPayload(anyLong(), anyLong(), anyLong(), anyString())) + .thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + long taskId = invocation.getArgument(1); + long resultId = invocation.getArgument(2); + String rowPayload = invocation.getArgument(3); + boolean duplicate = dbMembers.stream().anyMatch(m -> + Objects.equals(m.getDailyFileId(), dailyFileId) + && Objects.equals(m.getResultId(), resultId)); + if (duplicate) { + return false; + } + ShopDataCrawlDailyMemberEntity member = new ShopDataCrawlDailyMemberEntity(); + member.setId(memberIdSeq.incrementAndGet()); + member.setDailyFileId(dailyFileId); + member.setTaskId(taskId); + member.setResultId(resultId); + member.setRowPayload(rowPayload); + member.setCreatedAt(BUSINESS_TIME.plusMinutes(dbMembers.size())); + dbMembers.add(member); + return true; + }); + } + + private void captureDailyFilePersistence() { + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); + entity.setId(fileIdSeq.incrementAndGet()); + dbDailyFiles.add(entity); + return null; + }).when(dailyFileService).insert(any(ShopDataCrawlDailyFileEntity.class)); + lenient().doAnswer(invocation -> { + ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); + for (int i = 0; i < dbDailyFiles.size(); i++) { + ShopDataCrawlDailyFileEntity existing = dbDailyFiles.get(i); + if (Objects.equals(existing.getUserId(), entity.getUserId()) + && Objects.equals(existing.getShopKeyHash(), entity.getShopKeyHash()) + && Objects.equals(existing.getBusinessDate(), entity.getBusinessDate())) { + entity.setId(existing.getId()); + dbDailyFiles.set(i, copyDailyFile(entity)); + return null; + } + } + entity.setId(fileIdSeq.incrementAndGet()); + dbDailyFiles.add(copyDailyFile(entity)); + return null; + }).when(dailyFileService).update(any(ShopDataCrawlDailyFileEntity.class)); + lenient().when(dailyFileService.listMembers(anyLong())).thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + return dbMembers.stream() + .filter(m -> Objects.equals(m.getDailyFileId(), dailyFileId)) + .sorted(Comparator + .comparing(ShopDataCrawlDailyMemberEntity::getCreatedAt, + Comparator.nullsLast(Comparator.naturalOrder())) + .thenComparing(ShopDataCrawlDailyMemberEntity::getId, + Comparator.nullsLast(Comparator.naturalOrder()))) + .toList(); + }); + lenient().when(dailyFileService.containsResult(anyLong(), anyLong())).thenAnswer(invocation -> { + long dailyFileId = invocation.getArgument(0); + long resultId = invocation.getArgument(1); + return dbMembers.stream().anyMatch(m -> + Objects.equals(m.getDailyFileId(), dailyFileId) && Objects.equals(m.getResultId(), resultId)); + }); + lenient().when(dailyFileService.findMembersByResultId(anyLong())).thenAnswer(invocation -> { + long resultId = invocation.getArgument(0); + return dbMembers.stream().filter(m -> Objects.equals(m.getResultId(), resultId)).toList(); + }); + } + + private ShopDataCrawlDailyFileEntity findDailyFile(Long userId, String shopKeyHash) { + for (ShopDataCrawlDailyFileEntity f : dbDailyFiles) { + if (Objects.equals(f.getUserId(), userId) && Objects.equals(f.getShopKeyHash(), shopKeyHash)) { + return f; + } + } + return null; + } + + private ShopDataCrawlDailyFileEntity copyDailyFile(ShopDataCrawlDailyFileEntity source) { + if (source == null) { + return null; + } + ShopDataCrawlDailyFileEntity copy = new ShopDataCrawlDailyFileEntity(); + copy.setId(source.getId()); + copy.setUserId(source.getUserId()); + copy.setShopKeyHash(source.getShopKeyHash()); + copy.setShopKey(source.getShopKey()); + copy.setBusinessDate(source.getBusinessDate()); + copy.setLatestTaskId(source.getLatestTaskId()); + copy.setLatestResultId(source.getLatestResultId()); + copy.setResultFilename(source.getResultFilename()); + copy.setResultFileUrl(source.getResultFileUrl()); + copy.setResultFileSize(source.getResultFileSize()); + copy.setResultContentType(source.getResultContentType()); + copy.setRowCount(source.getRowCount()); + copy.setVersion(source.getVersion()); + copy.setLastSuccessAt(source.getLastSuccessAt()); + copy.setCreatedAt(source.getCreatedAt()); + copy.setUpdatedAt(source.getUpdatedAt()); + return copy; + } +}