EasyFlow-Tech-Dev/EasyFlow
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 恢复服务重启后的自动导入任务

Browse Source
This commit is contained in:
陈子默
2026-09-04 11:29:06 +08:00
parent c3673ece46
commit 788c8e5459
17 changed files with 2676 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/document/exception/DocumentParseBridgeException.java
View File

@@ -75,6 +75,10 @@ public class DocumentParseBridgeException extends RuntimeException {
return new DocumentParseBridgeException("task_failed", message);
}
public static DocumentParseBridgeException taskNotFound(String message, Throwable cause) {
return new DocumentParseBridgeException("task_not_found", message, cause);
}
public static DocumentParseBridgeException resultFetchFailed(String message, Throwable cause) {
return new DocumentParseBridgeException("result_fetch_failed", message, cause);
}

9
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/document/service/impl/DocumentParseBridgeServiceImpl.java
View File

@@ -5,6 +5,7 @@ import com.easyagents.document.core.entity.ParseResponse;
import com.easyagents.document.core.entity.ParseResult;
import com.easyagents.document.core.entity.ParseTaskInfo;
import com.easyagents.document.core.entity.ParseTaskStatus;
import com.easyagents.document.core.exception.DocumentAsyncTaskNotFoundException;
import com.easyagents.document.pdf.PdfDocumentParseService;
import com.easyagents.document.pptx.PptxDocumentParseService;
import com.easyagents.document.xlsx.XlsxDocumentParseService;
@@ -140,6 +141,8 @@ public class DocumentParseBridgeServiceImpl implements DocumentParseBridgeServic
return parseResultMapper.map(taskStatus);
} catch (DocumentParseBridgeException e) {
throw e;
} catch (DocumentAsyncTaskNotFoundException e) {
throw DocumentParseBridgeException.taskNotFound("异步文档解析任务不存在", e);
} catch (Exception e) {
throw DocumentParseBridgeException.taskFailed("查询异步文档解析任务状态失败", e);
}
@@ -175,6 +178,9 @@ public class DocumentParseBridgeServiceImpl implements DocumentParseBridgeServic
} catch (DocumentParseBridgeException e) {
LOG.error("桥接服务获取异步解析结果失败: providerTaskId={}", taskId, e);
throw e;
} catch (DocumentAsyncTaskNotFoundException e) {
LOG.warn("桥接服务异步解析执行已丢失: providerTaskId={}", taskId);
throw DocumentParseBridgeException.taskNotFound("异步文档解析任务不存在", e);
} catch (Exception e) {
LOG.error("桥接服务获取异步解析结果异常: providerTaskId={}", taskId, e);
throw DocumentParseBridgeException.resultFetchFailed("获取异步文档解析结果失败", e);
@@ -212,6 +218,9 @@ public class DocumentParseBridgeServiceImpl implements DocumentParseBridgeServic
} catch (DocumentParseBridgeException e) {
LOG.error("桥接服务查询异步解析任务状态失败: providerTaskId={}", taskId, e);
throw e;
} catch (DocumentAsyncTaskNotFoundException e) {
LOG.warn("桥接服务异步解析执行已丢失: providerTaskId={}", taskId);
throw DocumentParseBridgeException.taskNotFound("异步文档解析任务不存在", e);
} catch (Exception e) {
LOG.error("桥接服务查询异步解析任务状态异常: providerTaskId={}", taskId, e);
throw DocumentParseBridgeException.taskFailed("聚合查询异步文档解析任务信息失败", e);

25
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppService.java
View File

@@ -80,6 +80,7 @@ public class DocumentImportBatchAppService {
private final DocumentMapper documentMapper;
private final RedisLockExecutor redisLockExecutor;
private final DocumentImportBatchCircuitBreaker circuitBreaker;
private final DocumentImportRecoveryCoordinator recoveryCoordinator;
@Resource(name = "default")
private FileStorageService storageService;
@@ -97,6 +98,7 @@ public class DocumentImportBatchAppService {
* @param documentMapper 文档 Mapper
* @param redisLockExecutor 分布式锁执行器
* @param circuitBreaker 自动导入批次熔断器
* @param recoveryCoordinator 恢复令牌事务协调器
*/
public DocumentImportBatchAppService(DocumentImportBatchService batchService,
DocumentImportBatchItemService itemService,
@@ -107,7 +109,8 @@ public class DocumentImportBatchAppService {
DocumentImportBatchItemMapper itemMapper,
DocumentMapper documentMapper,
RedisLockExecutor redisLockExecutor,
DocumentImportBatchCircuitBreaker circuitBreaker) {
DocumentImportBatchCircuitBreaker circuitBreaker,
DocumentImportRecoveryCoordinator recoveryCoordinator) {
this.batchService = batchService;
this.itemService = itemService;
this.batchTracker = batchTracker;
@@ -118,6 +121,7 @@ public class DocumentImportBatchAppService {
this.documentMapper = documentMapper;
this.redisLockExecutor = redisLockExecutor;
this.circuitBreaker = circuitBreaker;
this.recoveryCoordinator = recoveryCoordinator;
}
/**
@@ -1246,13 +1250,10 @@ public class DocumentImportBatchAppService {
Date claimedAt = new Date();
Date leaseUntil = new Date(
claimedAt.getTime() + RECOVERY_DISPATCH_LEASE.toMillis());
if (batchMapper.claimRecoveryPending(
batchId, recoveryToken, leaseUntil, claimedAt) <= 0) {
return;
}
try {
DocumentImportBatch claimedBatch =
batchMapper.selectClaimedRecovery(batchId, recoveryToken);
recoveryCoordinator.claim(
batchId, recoveryToken, leaseUntil, claimedAt);
if (claimedBatch == null) {
LOG.info(
"批次恢复调度令牌已失效，旧持有者停止恢复: "
@@ -1284,9 +1285,8 @@ public class DocumentImportBatchAppService {
);
return;
}
int finalized = batchMapper.finalizeRecoveryPending(
batchId, recoveryToken, new Date());
if (finalized <= 0) {
if (!recoveryCoordinator.finalizeRecovery(
batchId, recoveryToken, new Date())) {
LOG.info(
"批次恢复待办已变更，当前实例跳过收尾: "
+ "batchId={}, recoveryToken={}",
@@ -1299,7 +1299,8 @@ public class DocumentImportBatchAppService {
if (!circuitBreaker.interruptRecoveryBatch(
batchId, recoveryToken, error)) {
LOG.info(
"批次恢复异常发生时令牌已失效，跳过旧持有者熔断: "
"批次恢复异常发生时令牌已失效或已有并发推进，"
+ "跳过当前持有者熔断: "
+ "batchId={}, recoveryToken={}",
batchId,
recoveryToken
@@ -1350,13 +1351,13 @@ public class DocumentImportBatchAppService {
Date renewedLeaseUntil = new Date(
nowMillis + RECOVERY_DISPATCH_LEASE.toMillis()
);
int renewed = batchMapper.renewRecoveryPendingLease(
boolean renewed = recoveryCoordinator.renew(
batchId,
recoveryToken,
renewedLeaseUntil,
now
);
if (renewed <= 0) {
if (!renewed) {
return false;
}
renewAfter.set(nowMillis + renewIntervalMillis);

84
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchCircuitBreaker.java
View File

@@ -9,7 +9,9 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import tech.easyflow.ai.entity.DocumentImportBatch;
import tech.easyflow.ai.entity.DocumentImportBatchItem;
import tech.easyflow.ai.entity.DocumentImportTask;
import tech.easyflow.ai.enums.DocumentImportBatchItemStatus;
import tech.easyflow.ai.enums.DocumentImportBatchStatus;
import tech.easyflow.ai.enums.DocumentImportMode;
import tech.easyflow.ai.enums.DocumentImportTaskStatus;
@@ -23,6 +25,8 @@ import java.sql.SQLRecoverableException;
import java.sql.SQLTimeoutException;
import java.sql.SQLTransientConnectionException;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.RejectedExecutionException;
/**
@@ -135,13 +139,18 @@ public class DocumentImportBatchCircuitBreaker {
if (recoveryToken == null || recoveryToken.isBlank()) {
throw new IllegalArgumentException("恢复调度令牌不能为空");
}
DocumentImportRecoveryException recoveryError =
findCause(error, DocumentImportRecoveryException.class);
return interruptBatch(
batchId,
resolveReason(error),
null,
null,
error,
recoveryToken
recoveryToken,
recoveryError == null
? List.of()
: recoveryError.getFailureFenceItemIds()
);
}
@@ -211,7 +220,7 @@ public class DocumentImportBatchCircuitBreaker {
String phase,
Throwable error) {
return interruptBatch(
batchId, reason, taskId, phase, error, null);
batchId, reason, taskId, phase, error, null, List.of());
}
/**
@@ -223,6 +232,7 @@ public class DocumentImportBatchCircuitBreaker {
* @param phase 触发任务阶段，可为空
* @param error 原始异常
* @param recoveryToken 恢复调度令牌，可为空
* @param recoveryFailureItemIds 零重排时的初始失败项围栏
* @return 批次是否已经停止运行；围栏失效时返回 {@code false}
*/
private boolean interruptBatch(BigInteger batchId,
@@ -230,11 +240,17 @@ public class DocumentImportBatchCircuitBreaker {
BigInteger taskId,
String phase,
Throwable error,
String recoveryToken) {
String recoveryToken,
List<BigInteger> recoveryFailureItemIds) {
if (batchId == null) {
return false;
}
DocumentImportBatch batch = batchMapper.selectOneById(batchId);
boolean fencedRecoveryFailure = recoveryToken != null
&& recoveryFailureItemIds != null
&& !recoveryFailureItemIds.isEmpty();
// 批次行是后续 task/item 状态收口的根锁；所有熔断路径先锁定
// batch，避免多表围栏更新与任务完成事务形成 task -> batch 逆序。
DocumentImportBatch batch = batchMapper.selectForUpdate(batchId);
if (batch == null) {
LOG.warn(
"忽略无法关联批次的自动导入熔断请求: taskId={}, batchId={}",
@@ -249,6 +265,39 @@ public class DocumentImportBatchCircuitBreaker {
}
Date now = new Date();
if (fencedRecoveryFailure) {
if (!Boolean.TRUE.equals(batch.getRecoveryPending())
|| !Objects.equals(recoveryToken, batch.getRecoveryToken())
|| batch.getRecoveryLeaseUntil() == null
|| !batch.getRecoveryLeaseUntil().after(now)) {
return false;
}
if (itemMapper.countActiveItems(batchId) > 0) {
LOG.info(
"恢复失败熔断发现批次已有活跃文件项，保留运行批次: "
+ "batchId={}",
batchId
);
return false;
}
for (BigInteger itemId : recoveryFailureItemIds) {
DocumentImportBatchItem item =
itemMapper.selectForUpdate(itemId);
if (item == null
|| !batchId.equals(item.getBatchId())
|| !DocumentImportBatchItemStatus.FAILED
.name().equals(item.getStatus())) {
LOG.info(
"恢复失败熔断发现等价并发推进，保留运行批次: "
+ "batchId={}, itemId={}, status={}",
batchId,
itemId,
item == null ? null : item.getStatus()
);
return false;
}
}
}
int interrupted;
if (recoveryToken != null) {
interrupted = batchMapper.interruptOwnedRecoveryBatch(
@@ -305,6 +354,14 @@ public class DocumentImportBatchCircuitBreaker {
* @return 稳定错误码与用户可见摘要
*/
private InterruptionReason resolveReason(Throwable error) {
DocumentImportRecoveryException recoveryError =
findCause(error, DocumentImportRecoveryException.class);
if (recoveryError != null) {
return new InterruptionReason(
recoveryError.getCode(),
recoveryError.getUserMessage()
);
}
if (containsRedisFailure(error)) {
return new InterruptionReason(
REDIS_UNAVAILABLE,
@@ -329,6 +386,25 @@ public class DocumentImportBatchCircuitBreaker {
);
}
/**
* 从异常链中查找指定类型。
*
* @param error 原始异常
* @param type 目标异常类型
* @param <T> 异常类型
* @return 匹配异常；不存在时返回 {@code null}
*/
private <T extends Throwable> T findCause(Throwable error, Class<T> type) {
Throwable current = error;
while (current != null) {
if (type.isInstance(current)) {
return type.cast(current);
}
current = current.getCause();
}
return null;
}
/**
* 判断异常链是否来自 Redis/Lettuce。
*

191
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTracker.java
View File

@@ -27,6 +27,18 @@ import java.util.Date;
@Service
public class DocumentImportBatchTracker {
/**
* 单文件恢复失败原因的事务内处理结果。
*/
public enum FailedRetryErrorOutcome {
/** 失败原因已写回，文件项仍等待后续人工恢复。 */
ERROR_RECORDED,
/** 等价并发操作已把文件项推进出失败状态。 */
ALREADY_ADVANCED,
/** 批次或文件项已不再允许当前恢复请求写回。 */
REJECTED
}
private final DocumentImportBatchService batchService;
private final DocumentImportBatchItemService itemService;
private final DocumentImportBatchMapper batchMapper;
@@ -64,6 +76,31 @@ public class DocumentImportBatchTracker {
return batch;
}
/**
* 在任务行变更前锁定所属批次，统一 batch -> task 锁序。
*
* <p>自动导入的任务生命周期由运行批次驱动，批次结束后不允许
* 旧任务继续推进。手动导入的上传批次可以先结束，用户之后再手动
* 执行分块和索引，因此只参与锁序，不以批次终态阻断后续任务。</p>
*
* @param batchId 批次 ID；无批次任务传入 {@code null}
* @return 批次存在且允许当前任务推进时为 {@code true}
*/
public boolean lockBatchForTaskMutation(BigInteger batchId) {
if (batchId == null) {
return true;
}
DocumentImportBatch batch = batchMapper.selectForUpdate(batchId);
if (batch == null) {
return false;
}
if (DocumentImportMode.AUTO.name().equals(batch.getImportMode())) {
return DocumentImportBatchStatus.RUNNING.name().equals(
batch.getStatus());
}
return DocumentImportMode.MANUAL.name().equals(batch.getImportMode());
}
/**
* 查询批次文件项。
*
@@ -107,10 +144,8 @@ public class DocumentImportBatchTracker {
if (itemId == null) {
return false;
}
DocumentImportBatchItem current = requireItem(itemId);
int attemptDelta = (status == DocumentImportBatchItemStatus.PENDING
|| status == DocumentImportBatchItemStatus.RUNNING)
&& DocumentImportBatchItemStatus.FAILED.name().equals(current.getStatus())
? 1
: 0;
return transitionItem(itemId, stage, status, errorSummary,
@@ -162,28 +197,34 @@ public class DocumentImportBatchTracker {
if (itemId == null) {
return false;
}
for (int attempt = 0; attempt < 3; attempt++) {
DocumentImportBatchItem current = requireItem(itemId);
LockedBatchItem locked = lockBatchThenItem(itemId);
if (DocumentImportBatchStatus.INTERRUPTED.name()
.equals(locked.batch.getStatus())) {
return false;
}
DocumentImportBatchItem current = locked.item;
DocumentImportBatchItemStatus currentStatus =
DocumentImportBatchItemStatus.valueOf(current.getStatus());
if (!isAllowedTransition(currentStatus, status)) {
return false;
}
String expectedStatus = current.getStatus();
Date now = new Date();
int effectiveAttemptDelta = currentStatus == DocumentImportBatchItemStatus.FAILED
? Math.max(0, attemptDelta)
: 0;
int updated = itemMapper.transitionStatus(
itemId,
expectedStatus,
current.getStatus(),
stage.name(),
status.name(),
errorSummary,
failureCode,
retryable,
Math.max(0, attemptDelta),
effectiveAttemptDelta,
now
);
if (updated <= 0) {
continue;
return false;
}
CounterDelta delta = CounterDelta.between(current, status, retryable);
if (!delta.isZero()) {
@@ -203,8 +244,6 @@ public class DocumentImportBatchTracker {
refreshBatch(current.getBatchId());
return true;
}
return false;
}
/**
* 校验文件项状态机，拒绝迟到任务覆盖终态。
@@ -243,7 +282,12 @@ public class DocumentImportBatchTracker {
*/
@org.springframework.transaction.annotation.Transactional
public void bindDocument(BigInteger itemId, BigInteger documentId) {
DocumentImportBatchItem item = requireItem(itemId);
LockedBatchItem locked = lockBatchThenItem(itemId);
if (DocumentImportBatchStatus.INTERRUPTED.name()
.equals(locked.batch.getStatus())) {
throw new BusinessException("导入批次已中断，请继续批次后重试");
}
DocumentImportBatchItem item = locked.item;
if (documentId.equals(item.getDocumentId())
&& DocumentImportBatchItemStatus.PENDING.name().equals(item.getStatus())) {
return;
@@ -252,6 +296,11 @@ public class DocumentImportBatchTracker {
boolean recoveringFailedItem =
DocumentImportBatchItemStatus.FAILED.name().equals(item.getStatus())
&& item.getDocumentId() == null;
if (recoveringFailedItem
&& !DocumentImportBatchStatus.RUNNING.name()
.equals(locked.batch.getStatus())) {
throw new BusinessException("导入批次状态已变化，请刷新后重试");
}
int updated = recoveringFailedItem
? itemMapper.bindFailedDocument(itemId, documentId, now)
: itemMapper.bindDocument(itemId, documentId, now);
@@ -268,6 +317,91 @@ public class DocumentImportBatchTracker {
refreshBatch(item.getBatchId());
}
/**
* 将引用已丢失文档的失败项重新绑定到恢复创建的新文档。
*
* @param itemId 文件项 ID
* @param missingDocumentId 已不存在的旧文档 ID
* @param documentId 恢复创建的新文档 ID
*/
@org.springframework.transaction.annotation.Transactional
public void replaceMissingDocument(BigInteger itemId,
BigInteger missingDocumentId,
BigInteger documentId) {
LockedBatchItem locked = lockBatchThenItem(itemId);
if (!DocumentImportBatchStatus.RUNNING.name()
.equals(locked.batch.getStatus())) {
throw new BusinessException("导入批次状态已变化，请刷新后重试");
}
DocumentImportBatchItem item = locked.item;
if (!DocumentImportBatchItemStatus.FAILED.name().equals(item.getStatus())
|| !java.util.Objects.equals(missingDocumentId, item.getDocumentId())) {
throw new BusinessException("导入文件状态已变化，请刷新后重试");
}
Date now = new Date();
if (itemMapper.replaceMissingDocument(
itemId, missingDocumentId, documentId, now) <= 0) {
throw new BusinessException("原文档状态已变化，请刷新后重试");
}
batchMapper.refreshCountersFromItems(item.getBatchId(), now);
refreshBatch(item.getBatchId());
}
/**
* 按批次到文件项的固定顺序取得失败项恢复权。
*
* @param itemId 文件项 ID
* @return 可恢复文件项；状态已变化时返回 null
*/
@org.springframework.transaction.annotation.Transactional
public DocumentImportBatchItem lockFailedItemForRetry(BigInteger itemId) {
if (itemId == null) {
return null;
}
LockedBatchItem locked = lockBatchThenItem(itemId);
if (!DocumentImportBatchStatus.RUNNING.name()
.equals(locked.batch.getStatus())
|| !DocumentImportBatchItemStatus.FAILED.name()
.equals(locked.item.getStatus())) {
return null;
}
return locked.item;
}
/**
* 在运行批次内保留单文件恢复失败原因。
*
* @param itemId 文件项 ID
* @param batchId 批次 ID
* @param errorSummary 错误摘要
* @return 失败原因写回、并发推进或拒绝结果
*/
@org.springframework.transaction.annotation.Transactional(
propagation = org.springframework.transaction.annotation.Propagation.REQUIRES_NEW)
public FailedRetryErrorOutcome updateFailedRetryError(
BigInteger itemId,
BigInteger batchId,
String errorSummary
) {
if (itemId == null || batchId == null) {
return FailedRetryErrorOutcome.REJECTED;
}
LockedBatchItem locked = lockBatchThenItem(itemId);
if (!batchId.equals(locked.batch.getId())
|| !DocumentImportBatchStatus.RUNNING.name()
.equals(locked.batch.getStatus())) {
return FailedRetryErrorOutcome.REJECTED;
}
if (!DocumentImportBatchItemStatus.FAILED.name()
.equals(locked.item.getStatus())) {
return FailedRetryErrorOutcome.ALREADY_ADVANCED;
}
return itemMapper.updateFailedRetryError(
itemId, batchId, errorSummary, new Date()) > 0
? FailedRetryErrorOutcome.ERROR_RECORDED
: FailedRetryErrorOutcome.REJECTED;
}
/**
* 原子完成文件上传并增量更新批次上传数。
*
@@ -280,7 +414,12 @@ public class DocumentImportBatchTracker {
public boolean completeUpload(BigInteger itemId,
String filePath,
String storageLocator) {
DocumentImportBatchItem item = requireItem(itemId);
LockedBatchItem locked = lockBatchThenItem(itemId);
if (DocumentImportBatchStatus.INTERRUPTED.name()
.equals(locked.batch.getStatus())) {
return false;
}
DocumentImportBatchItem item = locked.item;
if (DocumentImportBatchItemStatus.UPLOADED.name().equals(item.getStatus())) {
return filePath.equals(item.getFilePath())
&& storageLocator.equals(item.getStorageLocator());
@@ -304,6 +443,34 @@ public class DocumentImportBatchTracker {
return true;
}
/**
* 先锁批次再锁文件项，统一所有计数迁移路径的加锁顺序。
*
* @param itemId 文件项 ID
* @return 已锁定的批次与文件项
*/
private LockedBatchItem lockBatchThenItem(BigInteger itemId) {
DocumentImportBatchItem candidate = requireItem(itemId);
DocumentImportBatch batch = batchMapper.selectForUpdate(candidate.getBatchId());
if (batch == null) {
throw new BusinessException("导入批次不存在");
}
DocumentImportBatchItem item = itemMapper.selectForUpdate(itemId);
if (item == null) {
throw new BusinessException("导入文件不存在");
}
if (!java.util.Objects.equals(batch.getId(), item.getBatchId())) {
throw new IllegalStateException("导入文件批次归属已变化");
}
return new LockedBatchItem(batch, item);
}
private record LockedBatchItem(
DocumentImportBatch batch,
DocumentImportBatchItem item
) {
}
/**
* 记录文件项成功后需要清理的历史文档。
*

75
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportRecoveryCoordinator.java Normal file
View File

@@ -0,0 +1,75 @@
package tech.easyflow.ai.documentimport.task;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import tech.easyflow.ai.entity.DocumentImportBatch;
import tech.easyflow.ai.mapper.DocumentImportBatchMapper;
import java.math.BigInteger;
import java.util.Date;
/**
* 在独立短事务中维护批次恢复令牌。
*
* <p>继续操作通常从原事务的 {@code afterCommit} 回调启动。将领取、续租和
* 收尾放入独立事务，可保证恢复令牌在单文件重试事务开始前已经提交，并避免
* 复用刚完成提交但尚未解绑的事务资源。</p>
*/
@Service
public class DocumentImportRecoveryCoordinator {
private final DocumentImportBatchMapper batchMapper;
public DocumentImportRecoveryCoordinator(DocumentImportBatchMapper batchMapper) {
this.batchMapper = batchMapper;
}
/**
* 领取恢复待办并读取本次令牌对应的恢复参数。
*
* @param batchId 批次 ID
* @param recoveryToken 恢复令牌
* @param leaseUntil 租约到期时间
* @param claimedAt 领取时间
* @return 已领取批次；领取失败时返回 null
*/
@Transactional(propagation = Propagation.REQUIRES_NEW)
public DocumentImportBatch claim(BigInteger batchId,
String recoveryToken,
Date leaseUntil,
Date claimedAt) {
if (batchMapper.claimRecoveryPending(
batchId, recoveryToken, leaseUntil, claimedAt) <= 0) {
return null;
}
return batchMapper.selectClaimedRecovery(batchId, recoveryToken);
}
/**
* 续期当前恢复令牌。
*
* @return 当前令牌仍有效且续期成功时返回 true
*/
@Transactional(propagation = Propagation.REQUIRES_NEW)
public boolean renew(BigInteger batchId,
String recoveryToken,
Date leaseUntil,
Date renewedAt) {
return batchMapper.renewRecoveryPendingLease(
batchId, recoveryToken, leaseUntil, renewedAt) > 0;
}
/**
* 按真实计数收尾并清除恢复待办。
*
* @return 当前令牌仍有效且收尾成功时返回 true
*/
@Transactional(propagation = Propagation.REQUIRES_NEW)
public boolean finalizeRecovery(BigInteger batchId,
String recoveryToken,
Date finalizedAt) {
return batchMapper.finalizeRecoveryPending(
batchId, recoveryToken, finalizedAt) > 0;
}
}

71
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportRecoveryException.java Normal file
View File

@@ -0,0 +1,71 @@
package tech.easyflow.ai.documentimport.task;
import java.math.BigInteger;
import java.util.Collection;
import java.util.List;
/**
* 批次恢复未能重新排队任何失败项。
*
* <p>该异常只携带稳定错误码和用户可见摘要，供恢复令牌持有者
* 中断批次，避免“继续”请求在没有启动任何新任务时表现为成功。</p>
*/
public class DocumentImportRecoveryException extends RuntimeException {
private static final String ERROR_CODE = "document_import_recovery_failed";
private static final String USER_MESSAGE =
"未能重新排队任何失败文件，请查看文件错误后继续";
private final List<BigInteger> failureFenceItemIds;
/**
* 创建批次恢复失败异常。
*/
public DocumentImportRecoveryException() {
this(List.of());
}
/**
* 创建带初始失败项快照的批次恢复失败异常。
*
* @param failureFenceItemIds 领取恢复时仍为失败的文件项 ID
*/
public DocumentImportRecoveryException(
Collection<BigInteger> failureFenceItemIds
) {
super(USER_MESSAGE);
this.failureFenceItemIds = failureFenceItemIds == null
? List.of()
: failureFenceItemIds.stream()
.filter(java.util.Objects::nonNull)
.distinct()
.sorted()
.toList();
}
/**
* 获取稳定错误码。
*
* @return 稳定错误码
*/
public String getCode() {
return ERROR_CODE;
}
/**
* 获取用户可见摘要。
*
* @return 用户可见摘要
*/
public String getUserMessage() {
return USER_MESSAGE;
}
/**
* 获取用于原子熔断围栏的初始失败项。
*
* @return 按 ID 排序的不可变列表
*/
public List<BigInteger> getFailureFenceItemIds() {
return failureFenceItemIds;
}
}

674
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java
View File

File diff suppressed because it is too large Load Diff

112
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportBatchItemMapper.java
View File

@@ -17,22 +17,49 @@ import java.util.Date;
*/
public interface DocumentImportBatchItemMapper extends BaseMapper<DocumentImportBatchItem> {
String SELECT_COLUMNS = "id, batch_id AS batchId, knowledge_id AS knowledgeId, "
+ "document_id AS documentId, replaced_document_id AS replacedDocumentId, "
+ "client_file_key AS clientFileKey, file_name AS fileName, "
+ "relative_path AS relativePath, file_size AS fileSize, "
+ "file_path AS filePath, storage_locator AS storageLocator, "
+ "cleanup_pending AS cleanupPending, content_sha256 AS contentSha256, "
+ "stage, status, error_summary AS errorSummary, "
+ "failure_code AS failureCode, applied_strategy_code AS appliedStrategyCode, "
+ "strategy_snapshot_json AS strategySnapshotJson, retryable, "
+ "attempt_count AS attemptCount, created, created_by AS createdBy, "
+ "modified, modified_by AS modifiedBy";
/**
* 锁定运行批次中的失败项并取得本轮恢复权。
* 锁定并读取批次文件项。
*
* <p>同时锁定批次和文件项，使恢复任务创建与批次熔断串行化。</p>
* <p>调用方必须先锁定所属批次，再调用本方法，避免批次计数更新时
* 出现共享锁升级死锁。</p>
*
* @param itemId 文件项 ID
* @return 可恢复文件项；状态已变化时返回 null
* @return 文件项；不存在时返回 null
*/
@Select("SELECT item.* FROM tb_document_import_batch_item item "
+ "INNER JOIN tb_document_import_batch batch ON batch.id=item.batch_id "
+ "WHERE item.id=#{itemId} AND item.status='FAILED' "
+ "AND batch.status='RUNNING' FOR UPDATE")
DocumentImportBatchItem selectFailedForRetry(
@Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch_item "
+ "WHERE id=#{itemId} FOR UPDATE")
DocumentImportBatchItem selectForUpdate(
@Param("itemId") BigInteger itemId
);
/**
* 统计运行批次内已经由等价请求推进的活跃文件项。
*
* <p>调用方必须先锁定批次；所有文件项状态迁移遵循批次到文件项
* 的锁顺序，因此该当前状态检查到后续熔断之间不会新增活跃项。</p>
*
* @param batchId 批次 ID
* @return 活跃文件项数量
*/
@Select("SELECT COUNT(*) FROM tb_document_import_batch_item "
+ "WHERE batch_id=#{batchId} "
+ "AND status IN ('PENDING','UPLOADING','RUNNING','UPLOADED')")
int countActiveItems(
@Param("batchId") BigInteger batchId
);
/**
* 将中断批次中尚未结束的文件项统一收口为可恢复失败。
*
@@ -100,14 +127,13 @@ public interface DocumentImportBatchItemMapper extends BaseMapper<DocumentImport
* @param modified 修改时间
* @return 更新行数
*/
@Update("UPDATE tb_document_import_batch_item item "
+ "INNER JOIN tb_document_import_batch batch ON batch.id=item.batch_id SET "
+ "item.stage=#{stage}, item.status=#{status}, "
+ "item.error_summary=#{errorSummary}, item.failure_code=#{failureCode}, "
+ "item.retryable=#{retryable}, "
+ "item.attempt_count=item.attempt_count + #{attemptDelta}, "
+ "item.modified=#{modified} WHERE item.id=#{id} "
+ "AND item.status=#{expectedStatus} AND batch.status<>'INTERRUPTED'")
@Update("UPDATE tb_document_import_batch_item SET "
+ "stage=#{stage}, status=#{status}, "
+ "error_summary=#{errorSummary}, failure_code=#{failureCode}, "
+ "retryable=#{retryable}, "
+ "attempt_count=attempt_count + #{attemptDelta}, "
+ "modified=#{modified} WHERE id=#{id} "
+ "AND status=#{expectedStatus}")
int transitionStatus(
@Param("id") BigInteger id,
@Param("expectedStatus") String expectedStatus,
@@ -146,21 +172,46 @@ public interface DocumentImportBatchItemMapper extends BaseMapper<DocumentImport
* @param modified 修改时间
* @return 文件项仍属于运行批次且绑定成功时返回 1
*/
@Update("UPDATE tb_document_import_batch_item item "
+ "INNER JOIN tb_document_import_batch batch ON batch.id=item.batch_id SET "
+ "item.document_id=#{documentId}, item.stage='PARSE', "
+ "item.status='PENDING', item.error_summary=NULL, "
+ "item.failure_code=NULL, item.retryable=0, "
+ "item.attempt_count=COALESCE(item.attempt_count, 0) + 1, "
+ "item.modified=#{modified} "
+ "WHERE item.id=#{id} AND item.status='FAILED' "
+ "AND item.document_id IS NULL AND batch.status='RUNNING'")
@Update("UPDATE tb_document_import_batch_item SET "
+ "document_id=#{documentId}, stage='PARSE', "
+ "status='PENDING', error_summary=NULL, "
+ "failure_code=NULL, retryable=0, "
+ "attempt_count=COALESCE(attempt_count, 0) + 1, "
+ "modified=#{modified} "
+ "WHERE id=#{id} AND status='FAILED' "
+ "AND document_id IS NULL")
int bindFailedDocument(
@Param("id") BigInteger id,
@Param("documentId") BigInteger documentId,
@Param("modified") Date modified
);
/**
* 将引用已丢失文档的失败项绑定到恢复创建的新文档。
*
* @param id 文件项 ID
* @param missingDocumentId 已不存在的旧文档 ID
* @param documentId 恢复创建的新文档 ID
* @param modified 修改时间
* @return 文件项仍处于预期失败状态且旧文档确实不存在时返回 1
*/
@Update("UPDATE tb_document_import_batch_item SET "
+ "document_id=#{documentId}, stage='PARSE', "
+ "status='PENDING', error_summary=NULL, "
+ "failure_code=NULL, retryable=0, "
+ "attempt_count=COALESCE(attempt_count, 0) + 1, "
+ "modified=#{modified} "
+ "WHERE id=#{id} AND status='FAILED' "
+ "AND document_id=#{missingDocumentId} "
+ "AND NOT EXISTS (SELECT 1 FROM tb_document "
+ "WHERE id=#{missingDocumentId})")
int replaceMissingDocument(
@Param("id") BigInteger id,
@Param("missingDocumentId") BigInteger missingDocumentId,
@Param("documentId") BigInteger documentId,
@Param("modified") Date modified
);
/**
* 更新恢复失败项的业务错误，但不提前汇总批次终态。
*
@@ -170,12 +221,11 @@ public interface DocumentImportBatchItemMapper extends BaseMapper<DocumentImport
* @param modified 修改时间
* @return 文件项仍属于运行批次且更新成功时返回 1
*/
@Update("UPDATE tb_document_import_batch_item item "
+ "INNER JOIN tb_document_import_batch batch ON batch.id=item.batch_id SET "
+ "item.error_summary=#{errorSummary}, item.failure_code=NULL, "
+ "item.retryable=1, item.modified=#{modified} "
+ "WHERE item.id=#{id} AND item.batch_id=#{batchId} "
+ "AND item.status='FAILED' AND batch.status='RUNNING'")
@Update("UPDATE tb_document_import_batch_item SET "
+ "error_summary=#{errorSummary}, failure_code=NULL, "
+ "retryable=1, modified=#{modified} "
+ "WHERE id=#{id} AND batch_id=#{batchId} "
+ "AND status='FAILED'")
int updateFailedRetryError(
@Param("id") BigInteger id,
@Param("batchId") BigInteger batchId,

40
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportBatchMapper.java
View File

@@ -18,6 +18,38 @@ import java.util.List;
*/
public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatch> {
String SELECT_COLUMNS = "id, knowledge_id AS knowledgeId, "
+ "caller_type AS callerType, caller_id AS callerId, "
+ "idempotency_key_hash AS idempotencyKeyHash, "
+ "request_digest AS requestDigest, duplicate_policy AS duplicatePolicy, "
+ "requested_strategy_json AS requestedStrategyJson, "
+ "retry_generation AS retryGeneration, version, "
+ "import_mode AS importMode, status, total_count AS totalCount, "
+ "total_bytes AS totalBytes, completed_count AS completedCount, "
+ "processing_count AS processingCount, failed_count AS failedCount, "
+ "pending_count AS pendingCount, uploaded_count AS uploadedCount, "
+ "skipped_count AS skippedCount, cancelled_count AS cancelledCount, "
+ "retryable_failed_count AS retryableFailedCount, "
+ "interrupt_code AS interruptCode, interrupt_message AS interruptMessage, "
+ "interrupted_at AS interruptedAt, recovery_pending AS recoveryPending, "
+ "recovery_file_keys_json AS recoveryFileKeysJson, "
+ "recovery_token AS recoveryToken, "
+ "recovery_lease_until AS recoveryLeaseUntil, "
+ "started_at AS startedAt, finished_at AS finishedAt, created, "
+ "created_by AS createdBy, modified, modified_by AS modifiedBy";
/**
* 锁定并读取批次，统一批次计数与文件项状态更新的加锁顺序。
*
* @param batchId 批次 ID
* @return 批次；不存在时返回 null
*/
@Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
+ "WHERE id=#{batchId} FOR UPDATE")
DocumentImportBatch selectForUpdate(
@Param("batchId") BigInteger batchId
);
/**
* 锁定并读取调用者拥有的批次，保证后续重试基于最新状态。
*
@@ -26,7 +58,7 @@ public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatc
* @param callerId 调用者 ID
* @return 批次；不存在时返回 null
*/
@Select("SELECT * FROM tb_document_import_batch "
@Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
+ "WHERE id=#{batchId} AND caller_type=#{callerType} "
+ "AND caller_id=#{callerId} FOR UPDATE")
DocumentImportBatch selectOwnedForUpdate(
@@ -195,7 +227,7 @@ public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatc
* @param limit 最大批次数
* @return 待恢复批次
*/
@Select("SELECT * FROM tb_document_import_batch "
@Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
+ "WHERE status='RUNNING' AND recovery_pending=1 "
+ "AND (recovery_token IS NULL OR recovery_lease_until <= #{now}) "
+ "ORDER BY modified, id LIMIT #{limit}")
@@ -256,7 +288,7 @@ public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatc
* @param recoveryToken 恢复调度令牌
* @return 当前令牌持有的批次；令牌失效时返回空
*/
@Select("SELECT * FROM tb_document_import_batch "
@Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
+ "WHERE id=#{batchId} AND status='RUNNING' "
+ "AND recovery_pending=1 AND recovery_token=#{recoveryToken}")
DocumentImportBatch selectClaimedRecovery(
@@ -383,7 +415,7 @@ public interface DocumentImportBatchMapper extends BaseMapper<DocumentImportBatc
* @param limit 最大返回数量
* @return 待回收批次
*/
@Select("SELECT * FROM tb_document_import_batch "
@Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_batch "
+ "WHERE status IN ('UPLOADING','READY') "
+ "AND ((modified IS NOT NULL AND modified < #{incompleteCutoff}) "
+ "OR (modified IS NULL AND created < #{incompleteCutoff})) "

52
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportTaskMapper.java
View File

@@ -18,6 +18,26 @@ import java.util.List;
*/
public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask> {
String SELECT_COLUMNS = "id, document_id AS documentId, "
+ "knowledge_id AS knowledgeId, batch_id AS batchId, "
+ "batch_item_id AS batchItemId, phase, status, "
+ "provider_task_id AS providerTaskId, payload_json AS payloadJson, "
+ "error_summary AS errorSummary, failure_code AS failureCode, "
+ "attempt_no AS attemptNo, execution_token AS executionToken, "
+ "lease_until AS leaseUntil, version, started_at AS startedAt, "
+ "finished_at AS finishedAt, created, created_by AS createdBy, "
+ "modified, modified_by AS modifiedBy";
String QUALIFIED_SELECT_COLUMNS = "task.id, task.document_id AS documentId, "
+ "task.knowledge_id AS knowledgeId, task.batch_id AS batchId, "
+ "task.batch_item_id AS batchItemId, task.phase, task.status, "
+ "task.provider_task_id AS providerTaskId, task.payload_json AS payloadJson, "
+ "task.error_summary AS errorSummary, task.failure_code AS failureCode, "
+ "task.attempt_no AS attemptNo, task.execution_token AS executionToken, "
+ "task.lease_until AS leaseUntil, task.version, task.started_at AS startedAt, "
+ "task.finished_at AS finishedAt, task.created, task.created_by AS createdBy, "
+ "task.modified, task.modified_by AS modifiedBy";
/**
* 按任务阶段和批次轮转查询待投递任务，避免大批次独占投递窗口。
*
@@ -25,14 +45,16 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
* @param limit 最大任务数
* @return 公平排序后的待投递任务
*/
@Select("SELECT task.* FROM tb_document_import_task task JOIN ("
@Select("SELECT " + QUALIFIED_SELECT_COLUMNS
+ " FROM tb_document_import_task task JOIN ("
+ "SELECT task.id, ROW_NUMBER() OVER ("
+ "PARTITION BY task.phase, COALESCE(task.batch_id, task.id) "
+ "ORDER BY task.created, task.id) AS lane_row "
+ "FROM tb_document_import_task task "
+ "LEFT JOIN tb_document_import_batch batch ON batch.id=task.batch_id "
+ "WHERE task.status='PENDING' AND task.modified <= #{redispatchBefore} "
+ "AND (task.batch_id IS NULL OR batch.status='RUNNING')"
+ "AND (task.batch_id IS NULL OR batch.import_mode='MANUAL' "
+ "OR batch.status='RUNNING')"
+ ") ranked ON ranked.id=task.id "
+ "ORDER BY ranked.lane_row, task.created, task.id LIMIT #{limit}")
List<DocumentImportTask> selectPendingFairly(
@@ -52,11 +74,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
@Update("UPDATE tb_document_import_task SET modified=#{now}, "
+ "modified_by=#{operatorId}, version=COALESCE(version, 0) + 1 "
+ "WHERE id=#{id} AND status='PENDING' "
+ "AND modified <= #{redispatchBefore} "
+ "AND (batch_id IS NULL OR EXISTS ("
+ "SELECT 1 FROM tb_document_import_batch batch "
+ "WHERE batch.id=tb_document_import_task.batch_id "
+ "AND batch.status='RUNNING'))")
+ "AND modified <= #{redispatchBefore}")
int touchPendingForDispatch(
@Param("id") BigInteger id,
@Param("redispatchBefore") Date redispatchBefore,
@@ -72,7 +90,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
* @param limit 最大任务数
* @return 已失去执行租约的运行任务
*/
@Select("SELECT * FROM tb_document_import_task "
@Select("SELECT " + SELECT_COLUMNS + " FROM tb_document_import_task "
+ "WHERE status='RUNNING' AND ("
+ "(lease_until IS NOT NULL AND lease_until <= #{now}) OR "
+ "(lease_until IS NULL AND modified <= #{legacyCutoff})) "
@@ -99,11 +117,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
+ "started_at=COALESCE(started_at, #{now}), finished_at=NULL, "
+ "error_summary=NULL, failure_code=NULL, modified=#{now}, "
+ "modified_by=#{operatorId}, version=COALESCE(version, 0) + 1 "
+ "WHERE id=#{id} AND status='PENDING' "
+ "AND (batch_id IS NULL OR EXISTS ("
+ "SELECT 1 FROM tb_document_import_batch batch "
+ "WHERE batch.id=tb_document_import_task.batch_id "
+ "AND batch.status='RUNNING'))")
+ "WHERE id=#{id} AND status='PENDING'")
int claimPending(
@Param("id") BigInteger id,
@Param("executionToken") String executionToken,
@@ -126,11 +140,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
+ "modified=#{now}, modified_by=#{operatorId}, "
+ "version=COALESCE(version, 0) + 1 "
+ "WHERE id=#{id} AND status='RUNNING' "
+ "AND execution_token=#{executionToken} "
+ "AND (batch_id IS NULL OR EXISTS ("
+ "SELECT 1 FROM tb_document_import_batch batch "
+ "WHERE batch.id=tb_document_import_task.batch_id "
+ "AND batch.status='RUNNING'))")
+ "AND execution_token=#{executionToken}")
int renewLease(
@Param("id") BigInteger id,
@Param("executionToken") String executionToken,
@@ -156,11 +166,7 @@ public interface DocumentImportTaskMapper extends BaseMapper<DocumentImportTask>
+ "lease_until=NULL, finished_at=#{now}, modified=#{now}, "
+ "modified_by=#{operatorId}, version=COALESCE(version, 0) + 1 "
+ "WHERE id=#{id} AND status='RUNNING' "
+ "AND execution_token=#{executionToken} "
+ "AND (batch_id IS NULL OR EXISTS ("
+ "SELECT 1 FROM tb_document_import_batch batch "
+ "WHERE batch.id=tb_document_import_task.batch_id "
+ "AND batch.status='RUNNING'))")
+ "AND execution_token=#{executionToken}")
int finishOwned(
@Param("id") BigInteger id,
@Param("executionToken") String executionToken,

32
easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/document/service/impl/DocumentParseBridgeServiceImplTest.java
View File

@@ -6,6 +6,7 @@ import com.easyagents.document.core.entity.ParseResponse;
import com.easyagents.document.core.entity.ParseResult;
import com.easyagents.document.core.entity.ParseTaskInfo;
import com.easyagents.document.core.entity.ParseTaskStatus;
import com.easyagents.document.core.exception.DocumentAsyncTaskNotFoundException;
import com.easyagents.document.pdf.PdfDocumentParseService;
import com.easyagents.document.pptx.PptxDocumentParseService;
import com.easyagents.document.xlsx.XlsxDocumentParseService;
@@ -128,6 +129,33 @@ public class DocumentParseBridgeServiceImplTest {
Assert.assertEquals(0, pptxService.queryResultCallCount);
}
/**
* 验证本地 Office 内存任务丢失会转换为稳定桥接错误码。
*/
@Test
public void shouldExposeMissingOfficeTaskAsStableBridgeError() {
FakePptxDocumentParseService pptxService =
new FakePptxDocumentParseService();
pptxService.queryTaskInfoError =
new DocumentAsyncTaskNotFoundException("lost-task");
DocumentParseBridgeServiceImpl bridgeService =
buildBridgeService(null, pptxService, null, null);
try {
bridgeService.queryTaskInfo(
"lost-task",
buildSource(
"slides.pptx",
"application/vnd.openxmlformats-officedocument.presentationml.presentation"
)
);
Assert.fail("expected DocumentParseBridgeException");
} catch (DocumentParseBridgeException error) {
Assert.assertEquals("task_not_found", error.getCode());
Assert.assertSame(pptxService.queryTaskInfoError, error.getCause());
}
}
/**
* 验证缺少底层服务时抛出稳定错误码。
*/
@@ -418,6 +446,7 @@ public class DocumentParseBridgeServiceImplTest {
private int parseCallCount;
private int queryTaskInfoCallCount;
private int queryResultCallCount;
private RuntimeException queryTaskInfoError;
@Override
public ParseResponse parse(ParseRequest request) {
@@ -450,6 +479,9 @@ public class DocumentParseBridgeServiceImplTest {
@Override
public ParseTaskInfo queryTaskInfo(String taskId) {
queryTaskInfoCallCount++;
if (queryTaskInfoError != null) {
throw queryTaskInfoError;
}
throw new UnsupportedOperationException();
}
}

61
easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppServiceTest.java
View File

@@ -873,8 +873,8 @@ public class DocumentImportBatchAppServiceTest {
Mockito.anyString(),
Mockito.same(recoveryError)
);
Mockito.verify(context.batchMapper, Mockito.never())
.finalizeRecoveryPending(
Mockito.verify(context.recoveryCoordinator, Mockito.never())
.finalizeRecovery(
Mockito.any(), Mockito.anyString(), Mockito.any(Date.class));
}
@@ -892,8 +892,9 @@ public class DocumentImportBatchAppServiceTest {
Mockito.when(context.batchMapper.selectRecoveryPendingBatches(
Mockito.any(Date.class), Mockito.anyInt()
)).thenReturn(List.of(batch));
Mockito.when(context.batchMapper.selectClaimedRecovery(
Mockito.eq(batch.getId()), Mockito.anyString()
Mockito.when(context.recoveryCoordinator.claim(
Mockito.eq(batch.getId()), Mockito.anyString(),
Mockito.any(Date.class), Mockito.any(Date.class)
)).thenReturn(null);
int recovered = context.service.recoverPendingBatchRetries();
@@ -924,8 +925,9 @@ public class DocumentImportBatchAppServiceTest {
Mockito.when(context.batchMapper.selectRecoveryPendingBatches(
Mockito.any(Date.class), Mockito.anyInt()
)).thenReturn(List.of(batch));
Mockito.when(context.batchMapper.selectClaimedRecovery(
Mockito.eq(batch.getId()), Mockito.anyString()
Mockito.when(context.recoveryCoordinator.claim(
Mockito.eq(batch.getId()), Mockito.anyString(),
Mockito.any(Date.class), Mockito.any(Date.class)
)).thenReturn(claimedBatch);
int recovered = context.service.recoverPendingBatchRetries();
@@ -939,16 +941,14 @@ public class DocumentImportBatchAppServiceTest {
);
ArgumentCaptor<String> recoveryToken =
ArgumentCaptor.forClass(String.class);
Mockito.verify(context.batchMapper).claimRecoveryPending(
Mockito.verify(context.recoveryCoordinator).claim(
Mockito.eq(batch.getId()),
recoveryToken.capture(),
Mockito.any(Date.class),
Mockito.any(Date.class)
);
Mockito.verify(context.batchMapper).selectClaimedRecovery(
batch.getId(), recoveryToken.getValue());
Mockito.verify(context.batchMapper)
.finalizeRecoveryPending(
Mockito.verify(context.recoveryCoordinator)
.finalizeRecovery(
Mockito.eq(batch.getId()),
Mockito.eq(recoveryToken.getValue()),
Mockito.any(Date.class));
@@ -975,8 +975,8 @@ public class DocumentImportBatchAppServiceTest {
Assert.assertEquals(1, context.service.recoverPendingBatchRetries());
Mockito.verify(context.batchMapper, Mockito.never())
.finalizeRecoveryPending(
Mockito.verify(context.recoveryCoordinator, Mockito.never())
.finalizeRecovery(
Mockito.any(), Mockito.anyString(), Mockito.any(Date.class));
Mockito.verifyNoInteractions(context.circuitBreaker);
}
@@ -990,12 +990,12 @@ public class DocumentImportBatchAppServiceTest {
BigInteger batchId = BigInteger.valueOf(84);
String recoveryToken = "recovery-token";
AtomicLong now = new AtomicLong(1_000L);
Mockito.when(context.batchMapper.renewRecoveryPendingLease(
Mockito.when(context.recoveryCoordinator.renew(
Mockito.eq(batchId),
Mockito.eq(recoveryToken),
Mockito.any(Date.class),
Mockito.any(Date.class)
)).thenReturn(1, 0);
)).thenReturn(true, false);
BooleanSupplier guard = context.service.createRecoveryLeaseGuard(
batchId,
recoveryToken,
@@ -1004,8 +1004,8 @@ public class DocumentImportBatchAppServiceTest {
);
Assert.assertTrue(guard.getAsBoolean());
Mockito.verify(context.batchMapper, Mockito.never())
.renewRecoveryPendingLease(
Mockito.verify(context.recoveryCoordinator, Mockito.never())
.renew(
Mockito.any(),
Mockito.anyString(),
Mockito.any(Date.class),
@@ -1017,8 +1017,8 @@ public class DocumentImportBatchAppServiceTest {
now.set(121_000L);
Assert.assertFalse(guard.getAsBoolean());
Mockito.verify(context.batchMapper, Mockito.times(2))
.renewRecoveryPendingLease(
Mockito.verify(context.recoveryCoordinator, Mockito.times(2))
.renew(
Mockito.eq(batchId),
Mockito.eq(recoveryToken),
Mockito.any(Date.class),
@@ -1198,6 +1198,8 @@ public class DocumentImportBatchAppServiceTest {
RedisLockExecutor redisLockExecutor = Mockito.mock(RedisLockExecutor.class);
DocumentImportBatchCircuitBreaker circuitBreaker =
Mockito.mock(DocumentImportBatchCircuitBreaker.class);
DocumentImportRecoveryCoordinator recoveryCoordinator =
Mockito.mock(DocumentImportRecoveryCoordinator.class);
RedisLockExecutor.LockHandle lockHandle = Mockito.mock(RedisLockExecutor.LockHandle.class);
Mockito.when(redisLockExecutor.tryAcquire(
Mockito.anyString(), Mockito.any(), Mockito.any()
@@ -1213,21 +1215,16 @@ public class DocumentImportBatchAppServiceTest {
Mockito.when(batchMapper.selectOwnedForUpdate(
Mockito.any(), Mockito.anyString(), Mockito.any()
)).thenReturn(batch);
Mockito.when(batchMapper.claimRecoveryPending(
Mockito.any(),
Mockito.anyString(),
Mockito.any(Date.class),
Mockito.any(Date.class)
)).thenReturn(1);
Mockito.when(batchMapper.selectClaimedRecovery(
Mockito.any(), Mockito.anyString()
Mockito.when(recoveryCoordinator.claim(
Mockito.any(), Mockito.anyString(),
Mockito.any(Date.class), Mockito.any(Date.class)
)).thenReturn(batch);
Mockito.when(taskAppService.retryBatchFailures(
Mockito.any(), Mockito.anySet(), Mockito.any(BooleanSupplier.class)
)).thenReturn(true);
Mockito.when(batchMapper.finalizeRecoveryPending(
Mockito.when(recoveryCoordinator.finalizeRecovery(
Mockito.any(), Mockito.anyString(), Mockito.any(Date.class)
)).thenReturn(1);
)).thenReturn(true);
DocumentImportBatchAppService service = new DocumentImportBatchAppService(
batchService,
@@ -1239,12 +1236,13 @@ public class DocumentImportBatchAppServiceTest {
itemMapper,
documentMapper,
redisLockExecutor,
circuitBreaker
circuitBreaker,
recoveryCoordinator
);
return new TestContext(
service, batchService, itemService, batchTracker,
taskAppService, batchMapper, itemMapper, documentMapper,
circuitBreaker, lockHandle
circuitBreaker, recoveryCoordinator, lockHandle
);
}
@@ -1356,6 +1354,7 @@ public class DocumentImportBatchAppServiceTest {
DocumentImportBatchItemMapper itemMapper,
DocumentMapper documentMapper,
DocumentImportBatchCircuitBreaker circuitBreaker,
DocumentImportRecoveryCoordinator recoveryCoordinator,
RedisLockExecutor.LockHandle lockHandle
) {
}

169
easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchCircuitBreakerTest.java
View File

@@ -8,7 +8,9 @@ import org.springframework.dao.DataAccessResourceFailureException;
import org.springframework.jdbc.BadSqlGrammarException;
import org.springframework.data.redis.RedisSystemException;
import tech.easyflow.ai.entity.DocumentImportBatch;
import tech.easyflow.ai.entity.DocumentImportBatchItem;
import tech.easyflow.ai.entity.DocumentImportTask;
import tech.easyflow.ai.enums.DocumentImportBatchItemStatus;
import tech.easyflow.ai.enums.DocumentImportBatchStatus;
import tech.easyflow.ai.enums.DocumentImportMode;
import tech.easyflow.ai.enums.DocumentImportTaskPhase;
@@ -50,7 +52,7 @@ public class DocumentImportBatchCircuitBreakerTest {
DocumentImportTaskMapper taskMapper =
Mockito.mock(DocumentImportTaskMapper.class);
Mockito.when(taskMapper.selectOneById(taskId)).thenReturn(task);
Mockito.when(batchMapper.selectOneById(batchId)).thenReturn(batch);
Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
Mockito.when(batchMapper.interruptRunningBatchForActiveTask(
Mockito.eq(batchId),
Mockito.eq(taskId),
@@ -161,6 +163,7 @@ public class DocumentImportBatchCircuitBreakerTest {
DocumentImportTaskMapper taskMapper =
Mockito.mock(DocumentImportTaskMapper.class);
Mockito.when(batchMapper.selectOneById(batchId)).thenReturn(batch);
Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
Mockito.when(batchMapper.interruptOwnedRecoveryBatch(
Mockito.eq(batchId),
Mockito.eq("stale-token"),
@@ -183,6 +186,136 @@ public class DocumentImportBatchCircuitBreakerTest {
Mockito.any(Date.class), Mockito.any());
}
/**
* 验证零重排熔断在批次与失败项锁内完成，且未发生并发推进时生效。
*/
@Test
public void shouldInterruptRecoveryFailureWhenItemsRemainFailed() {
BigInteger batchId = BigInteger.valueOf(42);
BigInteger itemId = BigInteger.valueOf(43);
String recoveryToken = "current-token";
DocumentImportBatch batch = runningAutoBatch(batchId);
batch.setRecoveryPending(true);
batch.setRecoveryToken(recoveryToken);
batch.setRecoveryLeaseUntil(
new Date(System.currentTimeMillis() + 60_000));
DocumentImportBatchItem item = new DocumentImportBatchItem();
item.setId(itemId);
item.setBatchId(batchId);
item.setStatus(DocumentImportBatchItemStatus.FAILED.name());
DocumentImportBatchMapper batchMapper =
Mockito.mock(DocumentImportBatchMapper.class);
DocumentImportBatchItemMapper itemMapper =
Mockito.mock(DocumentImportBatchItemMapper.class);
DocumentImportTaskMapper taskMapper =
Mockito.mock(DocumentImportTaskMapper.class);
Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
Mockito.when(itemMapper.selectForUpdate(itemId)).thenReturn(item);
Mockito.when(batchMapper.interruptOwnedRecoveryBatch(
Mockito.eq(batchId), Mockito.eq(recoveryToken),
Mockito.eq("document_import_recovery_failed"),
Mockito.anyString(), Mockito.any(Date.class)
)).thenReturn(1);
DocumentImportBatchCircuitBreaker circuitBreaker =
new DocumentImportBatchCircuitBreaker(
batchMapper, itemMapper, taskMapper);
Assert.assertTrue(circuitBreaker.interruptRecoveryBatch(
batchId,
recoveryToken,
new DocumentImportRecoveryException(java.util.List.of(itemId))
));
org.mockito.InOrder order = Mockito.inOrder(batchMapper, itemMapper);
order.verify(batchMapper).selectForUpdate(batchId);
order.verify(itemMapper).countActiveItems(batchId);
order.verify(itemMapper).selectForUpdate(itemId);
order.verify(batchMapper).interruptOwnedRecoveryBatch(
Mockito.eq(batchId), Mockito.eq(recoveryToken),
Mockito.eq("document_import_recovery_failed"),
Mockito.anyString(), Mockito.any(Date.class));
}
/**
* 验证错误写回提交后出现的等价并发推进会阻止旧请求熔断整批。
*/
@Test
public void shouldKeepRunningBatchWhenRecoveryItemAdvanced() {
BigInteger batchId = BigInteger.valueOf(44);
BigInteger itemId = BigInteger.valueOf(45);
String recoveryToken = "current-token";
DocumentImportBatch batch = runningAutoBatch(batchId);
batch.setRecoveryPending(true);
batch.setRecoveryToken(recoveryToken);
batch.setRecoveryLeaseUntil(
new Date(System.currentTimeMillis() + 60_000));
DocumentImportBatchItem item = new DocumentImportBatchItem();
item.setId(itemId);
item.setBatchId(batchId);
item.setStatus(DocumentImportBatchItemStatus.PENDING.name());
DocumentImportBatchMapper batchMapper =
Mockito.mock(DocumentImportBatchMapper.class);
DocumentImportBatchItemMapper itemMapper =
Mockito.mock(DocumentImportBatchItemMapper.class);
Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
Mockito.when(itemMapper.countActiveItems(batchId)).thenReturn(1);
DocumentImportBatchCircuitBreaker circuitBreaker =
new DocumentImportBatchCircuitBreaker(
batchMapper,
itemMapper,
Mockito.mock(DocumentImportTaskMapper.class));
Assert.assertFalse(circuitBreaker.interruptRecoveryBatch(
batchId,
recoveryToken,
new DocumentImportRecoveryException(java.util.List.of(itemId))
));
Mockito.verify(batchMapper, Mockito.never())
.interruptOwnedRecoveryBatch(
Mockito.any(), Mockito.anyString(), Mockito.anyString(),
Mockito.anyString(), Mockito.any(Date.class));
Mockito.verify(itemMapper, Mockito.never()).selectForUpdate(itemId);
}
/**
* 验证失败快照外已有活跃项时，选择性恢复也不能熔断并发任务。
*/
@Test
public void shouldKeepRunningBatchWhenAnotherItemAdvancedBeforeSnapshot() {
BigInteger batchId = BigInteger.valueOf(46);
BigInteger failedItemId = BigInteger.valueOf(47);
String recoveryToken = "current-token";
DocumentImportBatch batch = runningAutoBatch(batchId);
batch.setRecoveryPending(true);
batch.setRecoveryToken(recoveryToken);
batch.setRecoveryLeaseUntil(
new Date(System.currentTimeMillis() + 60_000));
DocumentImportBatchMapper batchMapper =
Mockito.mock(DocumentImportBatchMapper.class);
DocumentImportBatchItemMapper itemMapper =
Mockito.mock(DocumentImportBatchItemMapper.class);
Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
Mockito.when(itemMapper.countActiveItems(batchId)).thenReturn(1);
DocumentImportBatchCircuitBreaker circuitBreaker =
new DocumentImportBatchCircuitBreaker(
batchMapper,
itemMapper,
Mockito.mock(DocumentImportTaskMapper.class));
Assert.assertFalse(circuitBreaker.interruptRecoveryBatch(
batchId,
recoveryToken,
new DocumentImportRecoveryException(
java.util.List.of(failedItemId))
));
Mockito.verify(itemMapper, Mockito.never())
.selectForUpdate(failedItemId);
Mockito.verify(batchMapper, Mockito.never())
.interruptOwnedRecoveryBatch(
Mockito.any(), Mockito.anyString(), Mockito.anyString(),
Mockito.anyString(), Mockito.any(Date.class));
}
/**
* 验证 SQL 语法错误保留为通用系统异常，避免伪装成数据库不可用。
*/
@@ -192,7 +325,7 @@ public class DocumentImportBatchCircuitBreakerTest {
DocumentImportBatch batch = runningAutoBatch(batchId);
DocumentImportBatchMapper batchMapper =
Mockito.mock(DocumentImportBatchMapper.class);
Mockito.when(batchMapper.selectOneById(batchId)).thenReturn(batch);
Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
Mockito.when(batchMapper.interruptRunningBatch(
Mockito.eq(batchId),
Mockito.eq("document_import_infrastructure_failure"),
@@ -222,7 +355,7 @@ public class DocumentImportBatchCircuitBreakerTest {
DocumentImportBatch batch = runningAutoBatch(batchId);
DocumentImportBatchMapper batchMapper =
Mockito.mock(DocumentImportBatchMapper.class);
Mockito.when(batchMapper.selectOneById(batchId)).thenReturn(batch);
Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
Mockito.when(batchMapper.interruptRunningBatch(
Mockito.eq(batchId),
Mockito.eq("database_unavailable"),
@@ -242,6 +375,36 @@ public class DocumentImportBatchCircuitBreakerTest {
));
}
/**
* 验证没有文件重新排队时使用稳定恢复失败语义中断批次。
*/
@Test
public void shouldExposeRecoveryFailureInsteadOfGenericInfrastructureError() {
BigInteger batchId = BigInteger.valueOf(71);
DocumentImportBatch batch = runningAutoBatch(batchId);
DocumentImportBatchMapper batchMapper =
Mockito.mock(DocumentImportBatchMapper.class);
Mockito.when(batchMapper.selectForUpdate(batchId)).thenReturn(batch);
Mockito.when(batchMapper.interruptRunningBatch(
Mockito.eq(batchId),
Mockito.eq("document_import_recovery_failed"),
Mockito.eq("未能重新排队任何失败文件，请查看文件错误后继续"),
Mockito.any(Date.class)
)).thenReturn(1);
DocumentImportBatchCircuitBreaker circuitBreaker =
new DocumentImportBatchCircuitBreaker(
batchMapper,
Mockito.mock(DocumentImportBatchItemMapper.class),
Mockito.mock(DocumentImportTaskMapper.class)
);
Assert.assertTrue(circuitBreaker.interruptBatch(
batchId,
new IllegalStateException(
"recovery failed", new DocumentImportRecoveryException())
));
}
/**
* 验证恢复与任务熔断 SQL 都包含对应所有权围栏。
*

162
easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchTrackerTest.java
View File

@@ -1,6 +1,8 @@
package tech.easyflow.ai.documentimport.task;
import org.junit.Test;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import tech.easyflow.ai.documentimport.DocumentImportBatchDtos;
import tech.easyflow.ai.entity.DocumentImportBatch;
import tech.easyflow.ai.entity.DocumentImportBatchItem;
@@ -13,6 +15,7 @@ import tech.easyflow.ai.mapper.DocumentImportBatchMapper;
import tech.easyflow.ai.service.DocumentImportBatchItemService;
import tech.easyflow.ai.service.DocumentImportBatchService;
import java.lang.reflect.Method;
import java.math.BigInteger;
import java.util.Date;
@@ -25,6 +28,7 @@ import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -62,6 +66,39 @@ public class DocumentImportBatchTrackerTest {
verify(batchService, never()).updateById(batch, false);
}
/**
* 验证任务变更始终先锁批次，但只有自动批次受运行态门禁约束。
*/
@Test
public void shouldAllowManualTaskAfterUploadBatchCompleted() {
BigInteger automaticBatchId = BigInteger.valueOf(101);
BigInteger manualBatchId = BigInteger.valueOf(102);
DocumentImportBatch automatic = new DocumentImportBatch();
automatic.setId(automaticBatchId);
automatic.setImportMode(DocumentImportMode.AUTO.name());
automatic.setStatus(DocumentImportBatchStatus.COMPLETED.name());
DocumentImportBatch manual = new DocumentImportBatch();
manual.setId(manualBatchId);
manual.setImportMode(DocumentImportMode.MANUAL.name());
manual.setStatus(DocumentImportBatchStatus.COMPLETED.name());
DocumentImportBatchMapper batchMapper =
mock(DocumentImportBatchMapper.class);
when(batchMapper.selectForUpdate(automaticBatchId))
.thenReturn(automatic);
when(batchMapper.selectForUpdate(manualBatchId)).thenReturn(manual);
DocumentImportBatchTracker tracker = new DocumentImportBatchTracker(
mock(DocumentImportBatchService.class),
mock(DocumentImportBatchItemService.class),
batchMapper,
mock(DocumentImportBatchItemMapper.class)
);
assertFalse(tracker.lockBatchForTaskMutation(automaticBatchId));
assertTrue(tracker.lockBatchForTaskMutation(manualBatchId));
verify(batchMapper).selectForUpdate(automaticBatchId);
verify(batchMapper).selectForUpdate(manualBatchId);
}
/**
* 验证兼容可重试计数与人工可继续的失败总数保持一致。
*/
@@ -111,6 +148,8 @@ public class DocumentImportBatchTrackerTest {
item.setStatus(DocumentImportBatchItemStatus.FAILED.name());
item.setRetryable(true);
when(itemService.getById(item.getId())).thenReturn(item);
when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
when(itemMapper.transitionStatus(
eq(item.getId()),
eq(DocumentImportBatchItemStatus.FAILED.name()),
@@ -143,6 +182,14 @@ public class DocumentImportBatchTrackerTest {
eq(-1),
any(Date.class)
);
org.mockito.InOrder lockOrder = inOrder(batchMapper, itemMapper);
lockOrder.verify(batchMapper).selectForUpdate(batch.getId());
lockOrder.verify(itemMapper).selectForUpdate(item.getId());
lockOrder.verify(itemMapper).transitionStatus(
eq(item.getId()), eq(DocumentImportBatchItemStatus.FAILED.name()),
eq(DocumentImportBatchItemStage.INDEX.name()),
eq(DocumentImportBatchItemStatus.PENDING.name()), eq(null), eq(null),
eq(false), eq(1), any(Date.class));
}
/**
@@ -167,6 +214,8 @@ public class DocumentImportBatchTrackerTest {
item.setRetryable(true);
BigInteger documentId = BigInteger.valueOf(99);
when(itemService.getById(item.getId())).thenReturn(item);
when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
when(itemMapper.bindFailedDocument(
eq(item.getId()), eq(documentId), any(Date.class)
)).thenReturn(1);
@@ -184,6 +233,35 @@ public class DocumentImportBatchTrackerTest {
anyInt(), anyInt(), anyInt(), any(Date.class));
}
/**
* 验证失败项只能在运行批次内绑定恢复创建的文档。
*/
@Test(expected = tech.easyflow.common.web.exceptions.BusinessException.class)
public void shouldRejectRecoveredDocumentOutsideRunningBatch() {
DocumentImportBatchService batchService =
mock(DocumentImportBatchService.class);
DocumentImportBatchItemService itemService =
mock(DocumentImportBatchItemService.class);
DocumentImportBatchMapper batchMapper =
mock(DocumentImportBatchMapper.class);
DocumentImportBatchItemMapper itemMapper =
mock(DocumentImportBatchItemMapper.class);
DocumentImportBatch batch = batch(
DocumentImportBatchStatus.PARTIAL_SUCCEEDED, 1);
DocumentImportBatchItem item = new DocumentImportBatchItem();
item.setId(BigInteger.TEN);
item.setBatchId(batch.getId());
item.setStatus(DocumentImportBatchItemStatus.FAILED.name());
when(itemService.getById(item.getId())).thenReturn(item);
when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
DocumentImportBatchTracker tracker =
new DocumentImportBatchTracker(
batchService, itemService, batchMapper, itemMapper);
tracker.bindDocument(item.getId(), BigInteger.valueOf(99));
}
/**
* 验证迟到任务不能把已完成文件重新改为处理中。
*/
@@ -198,6 +276,9 @@ public class DocumentImportBatchTrackerTest {
item.setBatchId(BigInteger.ONE);
item.setStatus(DocumentImportBatchItemStatus.COMPLETED.name());
when(itemService.getById(item.getId())).thenReturn(item);
DocumentImportBatch batch = batch(DocumentImportBatchStatus.RUNNING, 1);
when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
DocumentImportBatchTracker tracker =
new DocumentImportBatchTracker(batchService, itemService, batchMapper, itemMapper);
@@ -210,6 +291,87 @@ public class DocumentImportBatchTrackerTest {
any(), any(), any(), any(), any(), any(), anyBoolean(), anyInt(), any(Date.class));
}
/**
* 验证批次中断后，已到达的工作线程不能再迁移文件项。
*/
@Test
public void shouldFenceLateTransitionAfterBatchInterrupted() {
DocumentImportBatchService batchService = mock(DocumentImportBatchService.class);
DocumentImportBatchItemService itemService = mock(DocumentImportBatchItemService.class);
DocumentImportBatchMapper batchMapper = mock(DocumentImportBatchMapper.class);
DocumentImportBatchItemMapper itemMapper = mock(DocumentImportBatchItemMapper.class);
DocumentImportBatch batch = batch(DocumentImportBatchStatus.INTERRUPTED, 1);
DocumentImportBatchItem item = new DocumentImportBatchItem();
item.setId(BigInteger.TEN);
item.setBatchId(batch.getId());
item.setStatus(DocumentImportBatchItemStatus.PENDING.name());
when(itemService.getById(item.getId())).thenReturn(item);
when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
DocumentImportBatchTracker tracker =
new DocumentImportBatchTracker(batchService, itemService, batchMapper, itemMapper);
assertFalse(tracker.updateItem(item.getId(),
DocumentImportBatchItemStage.PARSE,
DocumentImportBatchItemStatus.RUNNING,
null));
verify(itemMapper, never()).transitionStatus(
any(), any(), any(), any(), any(), any(), anyBoolean(), anyInt(), any(Date.class));
}
/**
* 验证恢复失败原因在提交后回调内仍使用独立事务持久化。
*
* @throws Exception 方法不存在时抛出
*/
@Test
public void failedRetryErrorShouldUseIndependentTransaction()
throws Exception {
Method method = DocumentImportBatchTracker.class.getMethod(
"updateFailedRetryError",
BigInteger.class,
BigInteger.class,
String.class
);
Transactional transactional = method.getAnnotation(Transactional.class);
assertEquals(Propagation.REQUIRES_NEW, transactional.propagation());
}
/**
* 验证失败原因写回与等价并发推进在同一锁内完成判定。
*/
@Test
public void failedRetryErrorShouldRecognizeConcurrentAdvance() {
DocumentImportBatchService batchService =
mock(DocumentImportBatchService.class);
DocumentImportBatchItemService itemService =
mock(DocumentImportBatchItemService.class);
DocumentImportBatchMapper batchMapper =
mock(DocumentImportBatchMapper.class);
DocumentImportBatchItemMapper itemMapper =
mock(DocumentImportBatchItemMapper.class);
DocumentImportBatch batch = batch(DocumentImportBatchStatus.RUNNING, 1);
DocumentImportBatchItem item = new DocumentImportBatchItem();
item.setId(BigInteger.TEN);
item.setBatchId(batch.getId());
item.setStatus(DocumentImportBatchItemStatus.PENDING.name());
when(itemService.getById(item.getId())).thenReturn(item);
when(batchMapper.selectForUpdate(batch.getId())).thenReturn(batch);
when(itemMapper.selectForUpdate(item.getId())).thenReturn(item);
DocumentImportBatchTracker tracker =
new DocumentImportBatchTracker(
batchService, itemService, batchMapper, itemMapper);
assertEquals(
DocumentImportBatchTracker.FailedRetryErrorOutcome.ALREADY_ADVANCED,
tracker.updateFailedRetryError(
item.getId(), batch.getId(), "状态已变化"));
verify(itemMapper, never()).updateFailedRetryError(
any(), any(), any(), any(Date.class));
}
private DocumentImportBatch batch(DocumentImportBatchStatus status, int totalCount) {
DocumentImportBatch batch = new DocumentImportBatch();
batch.setId(BigInteger.ONE);

74
easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportRecoveryCoordinatorTest.java Normal file
View File

@@ -0,0 +1,74 @@
package tech.easyflow.ai.documentimport.task;
import org.junit.Assert;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import tech.easyflow.ai.entity.DocumentImportBatch;
import tech.easyflow.ai.mapper.DocumentImportBatchMapper;
import java.lang.reflect.Method;
import java.math.BigInteger;
import java.util.Date;
/**
* {@link DocumentImportRecoveryCoordinator} 恢复令牌事务测试。
*/
public class DocumentImportRecoveryCoordinatorTest {
@Test
public void claimShouldReturnOnlyCurrentTokenBatch() {
DocumentImportBatchMapper mapper =
Mockito.mock(DocumentImportBatchMapper.class);
DocumentImportRecoveryCoordinator coordinator =
new DocumentImportRecoveryCoordinator(mapper);
BigInteger batchId = BigInteger.valueOf(81);
String token = "token";
Date claimedAt = new Date(1_000L);
Date leaseUntil = new Date(121_000L);
DocumentImportBatch batch = new DocumentImportBatch();
batch.setId(batchId);
Mockito.when(mapper.claimRecoveryPending(
batchId, token, leaseUntil, claimedAt)).thenReturn(1);
Mockito.when(mapper.selectClaimedRecovery(batchId, token))
.thenReturn(batch);
Assert.assertSame(batch, coordinator.claim(
batchId, token, leaseUntil, claimedAt));
}
@Test
public void claimShouldSkipReadWhenTokenWasNotAcquired() {
DocumentImportBatchMapper mapper =
Mockito.mock(DocumentImportBatchMapper.class);
DocumentImportRecoveryCoordinator coordinator =
new DocumentImportRecoveryCoordinator(mapper);
Assert.assertNull(coordinator.claim(
BigInteger.ONE, "token", new Date(2_000L), new Date(1_000L)));
Mockito.verify(mapper, Mockito.never())
.selectClaimedRecovery(Mockito.any(), Mockito.anyString());
}
@Test
public void tokenMutationsShouldAlwaysUseIndependentTransactions()
throws Exception {
assertRequiresNew("claim", BigInteger.class, String.class,
Date.class, Date.class);
assertRequiresNew("renew", BigInteger.class, String.class,
Date.class, Date.class);
assertRequiresNew("finalizeRecovery", BigInteger.class,
String.class, Date.class);
}
private void assertRequiresNew(String methodName, Class<?>... parameterTypes)
throws Exception {
Method method = DocumentImportRecoveryCoordinator.class.getMethod(
methodName, parameterTypes);
Transactional transactional = method.getAnnotation(Transactional.class);
Assert.assertNotNull(transactional);
Assert.assertEquals(Propagation.REQUIRES_NEW,
transactional.propagation());
}
}

1073
easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppServiceTest.java
View File

File diff suppressed because it is too large Load Diff