task-29: 合并 scope 状态查询与更新，减少单 chunk 数据库往返
persistResultScope 复用接收前已加载的 scope，不再重复 selectOne —— 每个分片接收的 scope 查询从 2 次降为 1 次；空分片校验前移至任何 scope 查询之前，非法分片零数据库往返。任务级分布式锁保证预读 scope 在锁内不过期。
This commit is contained in:
+12
-8
@@ -1251,11 +1251,6 @@ public class ShopDataCrawlTaskService {
|
||||
int chunkTotal = incoming.getChunkTotal();
|
||||
validateChunkMetadata(chunkIndex, chunkTotal);
|
||||
|
||||
String scopeKey = resultChunkScopeKey(shopKey);
|
||||
String scopeHash = DigestUtil.sha256Hex(scopeKey);
|
||||
TaskScopeStateEntity scope = findResultScope(taskId, scopeHash);
|
||||
validateChunkTotal(scope == null ? null : scope.getChunkTotal(), chunkTotal);
|
||||
|
||||
List<ShopDataCrawlCountryResultDto> countryResults = copyCountryResults(incoming.getCountryResults());
|
||||
if (!hasProcessableChunkData(incoming.getCountryResults())) {
|
||||
throw new BusinessException("店铺数据抓取结果分片内容为空，拒绝接收");
|
||||
@@ -1263,6 +1258,11 @@ public class ShopDataCrawlTaskService {
|
||||
String payloadJson = writeJson(countryResults, "序列化店铺数据抓取结果分片失败");
|
||||
String payloadHash = DigestUtil.sha256Hex(payloadJson);
|
||||
|
||||
String scopeKey = resultChunkScopeKey(shopKey);
|
||||
String scopeHash = DigestUtil.sha256Hex(scopeKey);
|
||||
TaskScopeStateEntity scope = findResultScope(taskId, scopeHash);
|
||||
validateChunkTotal(scope == null ? null : scope.getChunkTotal(), chunkTotal);
|
||||
|
||||
ensureRustfsPayloadStorageEnabled();
|
||||
String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(
|
||||
MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
|
||||
@@ -1300,7 +1300,7 @@ public class ShopDataCrawlTaskService {
|
||||
}
|
||||
|
||||
try {
|
||||
persistResultScope(taskId, scopeKey, scopeHash, chunkTotal, receivedChunkCount);
|
||||
persistResultScope(scope, taskId, scopeKey, scopeHash, chunkTotal, receivedChunkCount);
|
||||
} catch (RuntimeException ex) {
|
||||
// 分片行与 scope 计数器是一致性单元：状态写入失败则回滚本次插入的分片行与 payload，
|
||||
// 客户端重试会走全新插入路径，计数器不会因重试重复计数或永久欠计。
|
||||
@@ -1380,12 +1380,16 @@ public class ShopDataCrawlTaskService {
|
||||
return chunkTotal != null && chunkTotal > 0 ? Math.min(chunkTotal, count) : count;
|
||||
}
|
||||
|
||||
private void persistResultScope(Long taskId,
|
||||
/**
|
||||
* 复用接收前已加载的 scope（Task 29：合并查询与更新，单 chunk 的 scope 往返 2 次降为 1 次）。
|
||||
* 任务级分布式锁串行化同一任务的接收，预读 scope 在锁内不会过期。
|
||||
*/
|
||||
private void persistResultScope(TaskScopeStateEntity scope,
|
||||
Long taskId,
|
||||
String scopeKey,
|
||||
String scopeHash,
|
||||
int chunkTotal,
|
||||
int receivedChunkCount) {
|
||||
TaskScopeStateEntity scope = findResultScope(taskId, scopeHash);
|
||||
validateChunkTotal(scope == null ? null : scope.getChunkTotal(), chunkTotal);
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
if (scope == null) {
|
||||
|
||||
+603
@@ -0,0 +1,603 @@
|
||||
package com.nanri.aiimage.modules.shopdatacrawl.service;
|
||||
|
||||
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.dto.ShopDataCrawlCountryResultDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlRowDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlShopPayloadDto;
|
||||
import com.nanri.aiimage.modules.shopdatacrawl.model.dto.ShopDataCrawlSubmitResultRequest;
|
||||
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.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 org.springframework.dao.DuplicateKeyException;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
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.anyLong;
|
||||
import static org.mockito.ArgumentMatchers.anyString;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.ArgumentMatchers.nullable;
|
||||
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 29：合并 scope 状态查询与更新，减少单 chunk 数据库往返。
|
||||
* persistResultChunk 在接收前已查询 scope（用于校验 chunk_total），
|
||||
* persistResultScope 不再重复 selectOne，而是复用同一份已加载的 scope
|
||||
* 直接 updateById / insert —— 每个分片接收的 scope 往返从 2 次降到 1 次。
|
||||
* 任务级分布式锁串行化同一任务的接收，预读的 scope 在锁内不会过期。
|
||||
*/
|
||||
@ExtendWith(MockitoExtension.class)
|
||||
class ShopDataCrawlScopeMergeTest {
|
||||
|
||||
private static final String MODULE_TYPE = "SHOP_DATA_CRAWL";
|
||||
private static final String SHOP_NAME = "Demo Shop";
|
||||
|
||||
@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);
|
||||
}
|
||||
|
||||
@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;
|
||||
|
||||
private final List<TaskChunkEntity> storedChunks = new ArrayList<>();
|
||||
private final List<TaskScopeStateEntity> storedScopes = new ArrayList<>();
|
||||
private final Map<String, String> rustfsPayloads = new LinkedHashMap<>();
|
||||
private FileTaskEntity task;
|
||||
private FileResultEntity result;
|
||||
private int nextPayloadId;
|
||||
private boolean insertFails;
|
||||
private boolean scopeUpdateFails;
|
||||
|
||||
@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);
|
||||
|
||||
storedChunks.clear();
|
||||
storedScopes.clear();
|
||||
rustfsPayloads.clear();
|
||||
nextPayloadId = 0;
|
||||
insertFails = false;
|
||||
scopeUpdateFails = false;
|
||||
|
||||
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.getAllShopMergedPayload(anyLong())).thenReturn(Map.of());
|
||||
lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())).thenReturn(List.of());
|
||||
lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())).thenReturn(null);
|
||||
lenient().when(excelAssemblyService.countRows(any())).thenReturn(1);
|
||||
|
||||
lenient().when(fileTaskMapper.selectById(anyLong())).thenAnswer(invocation -> {
|
||||
Long taskId = invocation.getArgument(0);
|
||||
return task != null && Objects.equals(taskId, task.getId()) ? task : null;
|
||||
});
|
||||
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
|
||||
lenient().when(fileResultMapper.selectById(anyLong())).thenAnswer(invocation -> {
|
||||
Long resultId = invocation.getArgument(0);
|
||||
return result != null && Objects.equals(resultId, result.getId()) ? result : null;
|
||||
});
|
||||
lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<FileResultEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
return result != null && Objects.equals(taskId, result.getTaskId()) ? List.of(result) : List.of();
|
||||
});
|
||||
lenient().when(fileResultMapper.updateById(any(FileResultEntity.class))).thenReturn(1);
|
||||
|
||||
lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> {
|
||||
String value = invocation.getArgument(0);
|
||||
return value == null ? "" : value.trim();
|
||||
});
|
||||
|
||||
configureTransientPayloadStorage();
|
||||
configureChunkMapper();
|
||||
configureScopeMapper();
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_029_chunk_merge_normal_default_path() {
|
||||
// 正常输入：两个分片顺序到达，每个接收只查询一次 scope（预取复用），
|
||||
// 首次 insert、后续 updateById，齐集完成任务。
|
||||
givenRunningTask(1201L, 2201L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
|
||||
service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002"))));
|
||||
|
||||
assertEquals(2, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertEquals(1, storedScopes.get(0).getCompleted());
|
||||
assertEquals(1, result.getSuccess());
|
||||
assertTrue(task.getResultJson().indexOf("B001") < task.getResultJson().indexOf("B002"));
|
||||
// 每次接收 1 次 scope 查询（预取复用，不再重复查询）+ 1 次状态写入；
|
||||
// 第 1 次提交后任务仍 RUNNING，tryFinalizeTask 补偿探针再查 1 次。
|
||||
verify(taskScopeStateMapper, times(3)).selectOne(any());
|
||||
verify(taskScopeStateMapper, times(1)).insert(any(TaskScopeStateEntity.class));
|
||||
verify(taskScopeStateMapper, times(1)).updateById(any(TaskScopeStateEntity.class));
|
||||
verify(taskChunkMapper, never()).selectCount(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_029_chunk_merge_normal_multiple_items() {
|
||||
// 批量场景：乱序 2/3、3/3、1/3，每个接收一次 scope 查询，结果按 chunk_index 稳定合并。
|
||||
givenRunningTask(1202L, 2202L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(2, 3, "DE", row("2026-07-25", "B002"))));
|
||||
service.submitResult(task.getId(), request(chunk(3, 3, "UK", row("2026-07-26", "B003"))));
|
||||
service.submitResult(task.getId(), request(chunk(1, 3, "FR", row("2026-07-27", "B001"))));
|
||||
|
||||
assertEquals(3, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertEquals(1, storedScopes.get(0).getCompleted());
|
||||
assertEquals(1, result.getSuccess());
|
||||
String json = task.getResultJson();
|
||||
assertTrue(json.indexOf("B001") < json.indexOf("B002"));
|
||||
assertTrue(json.indexOf("B002") < json.indexOf("B003"));
|
||||
verify(taskScopeStateMapper, times(5)).selectOne(any());
|
||||
verify(taskScopeStateMapper, times(1)).insert(any(TaskScopeStateEntity.class));
|
||||
verify(taskScopeStateMapper, times(2)).updateById(any(TaskScopeStateEntity.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_029_chunk_merge_normal_repeated_operation_is_idempotent() {
|
||||
// 重复执行：同一分片重试命中唯一键，counter 不重复推进；
|
||||
// 每个接收一次 scope 查询 + 一次状态写入，无重复 scope 行。
|
||||
givenRunningTask(1203L, 2203L);
|
||||
ShopDataCrawlSubmitResultRequest request = request(chunk(1, 2, "DE", row("2026-07-25", "B001")));
|
||||
|
||||
service.submitResult(task.getId(), request);
|
||||
service.submitResult(task.getId(), request);
|
||||
|
||||
assertEquals(1, storedScopes.size(), "scope 状态只有一行");
|
||||
assertEquals(1, storedScopes.get(0).getReceivedChunkCount(), "重试不重复计数");
|
||||
assertEquals(1, storedChunks.size());
|
||||
// 两次提交各 1 次 scope 查询，两次提交后任务均 RUNNING，finalize 探针各 1 次。
|
||||
verify(taskScopeStateMapper, times(4)).selectOne(any());
|
||||
verify(taskScopeStateMapper, times(1)).insert(any(TaskScopeStateEntity.class));
|
||||
verify(taskScopeStateMapper, times(1)).updateById(any(TaskScopeStateEntity.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_029_chunk_merge_boundary_empty_input() {
|
||||
// 空输入：无可处理数据的分片在 scope 查询前拒绝，无任何数据库往返与资源创建。
|
||||
givenRunningTask(1204L, 2204L);
|
||||
|
||||
ShopDataCrawlShopPayloadDto emptyItems = legacyChunk(false, "DE", null);
|
||||
emptyItems.setChunkIndex(1);
|
||||
emptyItems.setChunkTotal(1);
|
||||
emptyItems.setCountryResults(List.of(countryWithItems("DE", List.of())));
|
||||
|
||||
BusinessException error = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(emptyItems)));
|
||||
assertTrue(error.getMessage().contains("内容为空"));
|
||||
|
||||
assertEquals(0, storedChunks.size());
|
||||
assertEquals(0, rustfsPayloads.size());
|
||||
assertEquals(0, storedScopes.size());
|
||||
// 空分片在锁内最终化探针之前被拒，不触发任何 scope 查询。
|
||||
verify(taskScopeStateMapper, never()).selectOne(any());
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_029_chunk_merge_boundary_single_item() {
|
||||
// 单元素：单分片 1/1，一次 scope 查询 + 一次 insert 即齐集完成。
|
||||
givenRunningTask(1205L, 2205L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 1, "DE", row("2026-07-25", "B001"))));
|
||||
|
||||
assertEquals(1, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertEquals(1, storedScopes.get(0).getCompleted());
|
||||
assertEquals(1, result.getSuccess());
|
||||
verify(taskScopeStateMapper, times(1)).selectOne(any());
|
||||
verify(taskScopeStateMapper, times(1)).insert(any(TaskScopeStateEntity.class));
|
||||
verify(taskScopeStateMapper, never()).updateById(any(TaskScopeStateEntity.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_029_chunk_merge_boundary_limit_and_overflow() {
|
||||
// 上限/超限：5/5 全量到达 + 齐集前一次重放，每个接收一次 scope 查询；
|
||||
// 重放 counter 不推进、不新增 scope 行，最终收敛到 chunk_total。
|
||||
givenRunningTask(1206L, 2206L);
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 5, "UK", row("2026-07-25", "B001"))));
|
||||
service.submitResult(task.getId(), request(chunk(1, 5, "UK", row("2026-07-25", "B001"))));
|
||||
|
||||
assertEquals(1, storedScopes.get(0).getReceivedChunkCount(), "重放不重复计数");
|
||||
assertEquals(1, storedScopes.size());
|
||||
|
||||
String[] countries = {"UK", "DE", "FR", "ES", "IT"};
|
||||
String[] dates = {"2026-07-25", "2026-07-26", "2026-07-27", "2026-07-28", "2026-07-29"};
|
||||
for (int i = 2; i <= 5; i++) {
|
||||
service.submitResult(task.getId(), request(chunk(i, 5, countries[i - 1], row(dates[i - 1], "B00" + i))));
|
||||
}
|
||||
|
||||
assertEquals(5, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertEquals(1, storedScopes.get(0).getCompleted());
|
||||
assertEquals(1, result.getSuccess());
|
||||
assertEquals(1, storedScopes.size(), "scope 行始终只有一行");
|
||||
// 6 次接收各 1 次 scope 查询，5 次未齐集的 finalize 探针各 1 次（第 6 次齐集，完成路径不重复探针）。
|
||||
verify(taskScopeStateMapper, times(11)).selectOne(any());
|
||||
verify(taskScopeStateMapper, times(1)).insert(any(TaskScopeStateEntity.class));
|
||||
verify(taskScopeStateMapper, times(5)).updateById(any(TaskScopeStateEntity.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_029_chunk_merge_invalid_input_rejected() {
|
||||
// 非法参数：非正 chunk_index/chunk_total 在 scope 查询前拒绝；
|
||||
// 跨分片改 chunk_total 在预取校验处拒绝，不产生状态写入。
|
||||
givenRunningTask(1207L, 2207L);
|
||||
|
||||
BusinessException zeroIndex = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(0, 1, "DE", row("2026-07-25", "B001")))));
|
||||
assertTrue(zeroIndex.getMessage().contains("chunk_index"));
|
||||
|
||||
BusinessException zeroTotal = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(1, 0, "DE", row("2026-07-25", "B001")))));
|
||||
assertTrue(zeroTotal.getMessage().contains("chunk_total"));
|
||||
|
||||
verify(taskScopeStateMapper, never()).selectOne(any());
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
|
||||
|
||||
BusinessException changedTotal = assertThrows(BusinessException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(2, 3, "UK", row("2026-07-26", "B002")))));
|
||||
assertTrue(changedTotal.getMessage().contains("chunk_total"));
|
||||
|
||||
assertEquals(1, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertEquals(1, storedChunks.size());
|
||||
// 3 次查询：chunk(1,2) 预取 + 其 RUNNING finalize 探针 + changedTotal 提交的预取（随后在校验处拒绝）。
|
||||
verify(taskScopeStateMapper, times(3)).selectOne(any());
|
||||
verify(taskScopeStateMapper, times(1)).insert(any(TaskScopeStateEntity.class));
|
||||
verify(taskScopeStateMapper, never()).updateById(any(TaskScopeStateEntity.class));
|
||||
}
|
||||
|
||||
@Test
|
||||
void test_task_029_chunk_merge_dependency_failure_releases_resources() {
|
||||
// 依赖失败：分片插入异常 → payload 清理、无状态写入，恢复后重试成功；
|
||||
// scope 更新失败 → 回滚本次插入的分片行与 payload，恢复后重试为全新插入并完成。
|
||||
givenRunningTask(1208L, 2208L);
|
||||
|
||||
insertFails = true;
|
||||
assertThrows(RuntimeException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001")))));
|
||||
assertEquals(0, storedChunks.size());
|
||||
assertEquals(0, rustfsPayloads.size());
|
||||
assertEquals(0, storedScopes.size(), "插入失败不产生 scope 行");
|
||||
insertFails = false;
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(1, 2, "DE", row("2026-07-25", "B001"))));
|
||||
assertEquals(1, storedScopes.get(0).getReceivedChunkCount());
|
||||
|
||||
scopeUpdateFails = true;
|
||||
assertThrows(RuntimeException.class, () ->
|
||||
service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002")))));
|
||||
assertEquals(1, storedChunks.size(), "状态写入失败回滚分片行");
|
||||
assertEquals(1, rustfsPayloads.size(), "状态写入失败清理 payload");
|
||||
assertEquals(1, storedScopes.get(0).getReceivedChunkCount(), "状态写入失败计数器未推进");
|
||||
scopeUpdateFails = false;
|
||||
|
||||
service.submitResult(task.getId(), request(chunk(2, 2, "UK", row("2026-07-26", "B002"))));
|
||||
|
||||
assertEquals(1, result.getSuccess(), "恢复后重试成功");
|
||||
assertEquals(2, storedScopes.get(0).getReceivedChunkCount());
|
||||
assertEquals(1, storedScopes.get(0).getCompleted());
|
||||
// 5 次 scope 查询：4 次接收各 1 次（含 1 次插入失败、1 次状态写入失败）+ chunk1 成功后的 finalize 探针。
|
||||
verify(taskScopeStateMapper, times(5)).selectOne(any());
|
||||
verify(taskScopeStateMapper, times(1)).insert(any(TaskScopeStateEntity.class));
|
||||
verify(taskScopeStateMapper, times(2)).updateById(any(TaskScopeStateEntity.class));
|
||||
}
|
||||
|
||||
private void configureTransientPayloadStorage() {
|
||||
lenient().when(transientPayloadStorageService.isSharedWriteEnabled()).thenReturn(true);
|
||||
lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(
|
||||
anyString(), anyLong(), anyString(), any(), anyString())).thenAnswer(invocation -> {
|
||||
String pointer = "rustfs:payload-" + (++nextPayloadId);
|
||||
rustfsPayloads.put(pointer, invocation.getArgument(4));
|
||||
return pointer;
|
||||
});
|
||||
lenient().when(transientPayloadStorageService.extractPointer(any())).thenAnswer(invocation -> {
|
||||
String value = invocation.getArgument(0);
|
||||
if (value == null) {
|
||||
return null;
|
||||
}
|
||||
return value.startsWith("rustfs:") || value.startsWith("local:") || value.startsWith("oss:")
|
||||
? value : null;
|
||||
});
|
||||
lenient().when(transientPayloadStorageService.resolvePayload(any(), any())).thenAnswer(invocation -> {
|
||||
String pointer = invocation.getArgument(0);
|
||||
String payload = rustfsPayloads.get(pointer);
|
||||
if (payload == null) {
|
||||
throw new IllegalStateException("missing test RustFS payload: " + pointer);
|
||||
}
|
||||
return payload;
|
||||
});
|
||||
lenient().doAnswer(invocation -> {
|
||||
rustfsPayloads.remove(invocation.getArgument(0));
|
||||
return null;
|
||||
}).when(transientPayloadStorageService).deletePayloadIfPresent(any());
|
||||
}
|
||||
|
||||
private void configureChunkMapper() {
|
||||
lenient().when(taskChunkMapper.insert(any(TaskChunkEntity.class))).thenAnswer(invocation -> {
|
||||
if (insertFails) {
|
||||
throw new RuntimeException("db down");
|
||||
}
|
||||
TaskChunkEntity chunk = invocation.getArgument(0);
|
||||
boolean duplicate = storedChunks.stream().anyMatch(existing ->
|
||||
Objects.equals(existing.getTaskId(), chunk.getTaskId())
|
||||
&& Objects.equals(existing.getScopeHash(), chunk.getScopeHash())
|
||||
&& Objects.equals(existing.getChunkIndex(), chunk.getChunkIndex()));
|
||||
if (duplicate) {
|
||||
throw new DuplicateKeyException("duplicate chunk key: " + chunk.getScopeHash() + "/" + chunk.getChunkIndex());
|
||||
}
|
||||
chunk.setId((long) storedChunks.size() + 1L);
|
||||
storedChunks.add(chunk);
|
||||
return 1;
|
||||
});
|
||||
lenient().when(taskChunkMapper.selectOne(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
Integer chunkIndex = queryInteger(query);
|
||||
return storedChunks.stream()
|
||||
.filter(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex))
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
});
|
||||
lenient().when(taskChunkMapper.selectList(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
return storedChunks.stream()
|
||||
.filter(chunk -> matchesChunk(chunk, taskId, scopeHash, null))
|
||||
.sorted(Comparator.comparing(TaskChunkEntity::getChunkIndex))
|
||||
.toList();
|
||||
});
|
||||
lenient().when(taskChunkMapper.delete(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskChunkEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
Integer chunkIndex = queryInteger(query);
|
||||
int before = storedChunks.size();
|
||||
storedChunks.removeIf(chunk -> matchesChunk(chunk, taskId, scopeHash, chunkIndex));
|
||||
return before - storedChunks.size();
|
||||
});
|
||||
}
|
||||
|
||||
private void configureScopeMapper() {
|
||||
lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> {
|
||||
TaskScopeStateEntity scope = invocation.getArgument(0);
|
||||
scope.setId((long) storedScopes.size() + 1L);
|
||||
storedScopes.add(scope);
|
||||
return 1;
|
||||
});
|
||||
lenient().when(taskScopeStateMapper.selectOne(any())).thenAnswer(invocation -> {
|
||||
LambdaQueryWrapper<TaskScopeStateEntity> query = invocation.getArgument(0);
|
||||
Long taskId = queryLong(query);
|
||||
String scopeHash = queryScopeHash(query);
|
||||
return storedScopes.stream()
|
||||
.filter(scope -> Objects.equals(taskId, scope.getTaskId()))
|
||||
.filter(scope -> scopeHash == null || Objects.equals(scopeHash, scope.getScopeHash()))
|
||||
.findFirst()
|
||||
.map(ShopDataCrawlScopeMergeTest::copyScope)
|
||||
.orElse(null);
|
||||
});
|
||||
lenient().when(taskScopeStateMapper.updateById(any(TaskScopeStateEntity.class))).thenAnswer(invocation -> {
|
||||
if (scopeUpdateFails) {
|
||||
throw new RuntimeException("scope update down");
|
||||
}
|
||||
TaskScopeStateEntity updated = invocation.getArgument(0);
|
||||
for (int i = 0; i < storedScopes.size(); i++) {
|
||||
if (Objects.equals(storedScopes.get(i).getId(), updated.getId())) {
|
||||
storedScopes.set(i, copyScope(updated));
|
||||
}
|
||||
}
|
||||
return 1;
|
||||
});
|
||||
}
|
||||
|
||||
private static TaskScopeStateEntity copyScope(TaskScopeStateEntity source) {
|
||||
TaskScopeStateEntity copy = new TaskScopeStateEntity();
|
||||
copy.setId(source.getId());
|
||||
copy.setTaskId(source.getTaskId());
|
||||
copy.setModuleType(source.getModuleType());
|
||||
copy.setScopeKey(source.getScopeKey());
|
||||
copy.setScopeHash(source.getScopeHash());
|
||||
copy.setChunkTotal(source.getChunkTotal());
|
||||
copy.setReceivedChunkCount(source.getReceivedChunkCount());
|
||||
copy.setCompleted(source.getCompleted());
|
||||
copy.setLastChunkAt(source.getLastChunkAt());
|
||||
copy.setLastError(source.getLastError());
|
||||
copy.setCreatedAt(source.getCreatedAt());
|
||||
copy.setUpdatedAt(source.getUpdatedAt());
|
||||
return copy;
|
||||
}
|
||||
|
||||
private boolean matchesChunk(TaskChunkEntity chunk, Long taskId, String scopeHash, Integer chunkIndex) {
|
||||
return (taskId == null || Objects.equals(taskId, chunk.getTaskId()))
|
||||
&& (scopeHash == null || Objects.equals(scopeHash, chunk.getScopeHash()))
|
||||
&& (chunkIndex == null || Objects.equals(chunkIndex, chunk.getChunkIndex()));
|
||||
}
|
||||
|
||||
private Long queryLong(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(Long.class::isInstance)
|
||||
.map(Long.class::cast)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private Integer queryInteger(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(Integer.class::isInstance)
|
||||
.map(Integer.class::cast)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private String queryScopeHash(LambdaQueryWrapper<?> query) {
|
||||
query.getSqlSegment();
|
||||
return query.getParamNameValuePairs().values().stream()
|
||||
.filter(String.class::isInstance)
|
||||
.map(String.class::cast)
|
||||
.filter(value -> value.length() == 64)
|
||||
.findFirst()
|
||||
.orElse(null);
|
||||
}
|
||||
|
||||
private void givenRunningTask(long taskId, long resultId) {
|
||||
task = new FileTaskEntity();
|
||||
task.setId(taskId);
|
||||
task.setUserId(7L);
|
||||
task.setModuleType(MODULE_TYPE);
|
||||
task.setStatus("RUNNING");
|
||||
task.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}");
|
||||
task.setResultJson("[]");
|
||||
task.setCreatedAt(LocalDateTime.now());
|
||||
task.setUpdatedAt(LocalDateTime.now());
|
||||
|
||||
result = new FileResultEntity();
|
||||
result.setId(resultId);
|
||||
result.setTaskId(taskId);
|
||||
result.setUserId(7L);
|
||||
result.setModuleType(MODULE_TYPE);
|
||||
result.setSourceFilename(SHOP_NAME);
|
||||
result.setSourceFileUrl("shop-1");
|
||||
result.setSuccess(-1);
|
||||
result.setCreatedAt(LocalDateTime.now());
|
||||
}
|
||||
|
||||
private ShopDataCrawlSubmitResultRequest request(ShopDataCrawlShopPayloadDto payload) {
|
||||
ShopDataCrawlSubmitResultRequest request = new ShopDataCrawlSubmitResultRequest();
|
||||
request.setShops(List.of(payload));
|
||||
return request;
|
||||
}
|
||||
|
||||
private ShopDataCrawlShopPayloadDto chunk(int chunkIndex,
|
||||
int chunkTotal,
|
||||
String country,
|
||||
ShopDataCrawlRowDto row) {
|
||||
ShopDataCrawlShopPayloadDto payload = legacyChunk(false, country, row);
|
||||
payload.setChunkIndex(chunkIndex);
|
||||
payload.setChunkTotal(chunkTotal);
|
||||
return payload;
|
||||
}
|
||||
|
||||
private ShopDataCrawlShopPayloadDto legacyChunk(boolean shopDone,
|
||||
String country,
|
||||
ShopDataCrawlRowDto row) {
|
||||
ShopDataCrawlShopPayloadDto payload = new ShopDataCrawlShopPayloadDto();
|
||||
payload.setShopName(SHOP_NAME);
|
||||
if (row != null) {
|
||||
payload.setCountryResults(List.of(country(country, row)));
|
||||
}
|
||||
payload.setShopDone(shopDone);
|
||||
return payload;
|
||||
}
|
||||
|
||||
private ShopDataCrawlCountryResultDto country(String country, ShopDataCrawlRowDto row) {
|
||||
return countryWithItems(country, List.of(row));
|
||||
}
|
||||
|
||||
private ShopDataCrawlCountryResultDto countryWithItems(String country, List<ShopDataCrawlRowDto> items) {
|
||||
ShopDataCrawlCountryResultDto result = new ShopDataCrawlCountryResultDto();
|
||||
result.setCountry(country);
|
||||
result.setItems(items);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ShopDataCrawlRowDto row(String date, String asin) {
|
||||
ShopDataCrawlRowDto row = new ShopDataCrawlRowDto();
|
||||
row.setDate(date);
|
||||
row.setAsin(asin);
|
||||
row.setCommodityImage("https://m.media-amazon.com/images/I/" + asin + ".jpg");
|
||||
row.setInventorySales("10");
|
||||
row.setSalesRank("20");
|
||||
row.setPageViews("30");
|
||||
row.setUnitsSold("40");
|
||||
row.setPrice("50");
|
||||
row.setRecommendedOffer("60");
|
||||
return row;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user