super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

task-36

Browse Source
This commit is contained in:
黄自达
2026-08-29 22:05:30 +08:00
parent f2546fc152
commit 93ccb27b40
3 changed files with 680 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/DailyStateConflictException.java
+13
View File
@@ -0,0 +1,13 @@
package com.nanri.aiimage.modules.shopdatacrawl.service;
/**
* Task 36：每日累计文件归档提交阶段的版本 CAS 冲突信号。
* 由提交阶段捕获并触发整次归档重试（释放店铺级锁后按最新状态重新组装），
* 区别于需要调用方直接失败的运行时异常。
*/
class DailyStateConflictException extends RuntimeException {
DailyStateConflictException() {
super("daily file state conflict");
}
}
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java
+66 -33
View File
@@ -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<ShopDataCrawlDailyFileEntity> olderFiles = dailyFileService.findOlder(
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlDailyFileLockTest.java
+601
View File
@@ -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<FileResultEntity> dbResultRows = new ArrayList<>();
/** 内存中的 daily_file 表。 */
private final List<ShopDataCrawlDailyFileEntity> dbDailyFiles = new ArrayList<>();
/** 内存中的 daily_member 表。 */
private final List<ShopDataCrawlDailyMemberEntity> 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<TaskDistributedLockService.LockHandle> lastLock = new AtomicReference<>();
/** 本次测试获取到的全部店铺级锁句柄（验证每个临界区取到的锁都被释放）。 */
private final List<TaskDistributedLockService.LockHandle> 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<FileResultEntity> wrapper = invocation.getArgument(0);
if (wrapper instanceof LambdaQueryWrapper<?> query
&& query.getSqlSegment() != null && query.getSqlSegment().contains("taskId")) {
List<FileResultEntity> 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<ShopDataCrawlDailyMemberEntity> 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<FileResultEntity> 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;
}
}