feat(file): expedite pending export retries
This commit is contained in:
@@ -19,6 +19,7 @@ public final class ExportTaskDtos {
|
||||
String id, String taskNo, String exportType, String format, String taskName,
|
||||
String status, Long totalRows, long processedRows, String fileName,
|
||||
String contentType, Long sizeBytes, String failureMessage,
|
||||
int attemptCount, Integer maxAttempts, LocalDateTime nextRetryAt,
|
||||
String requestedBy, String requestedUsername, LocalDateTime requestedAt,
|
||||
LocalDateTime startedAt, LocalDateTime completedAt, LocalDateTime expiresAt) { }
|
||||
}
|
||||
|
||||
@@ -33,6 +33,13 @@ public interface FileExportOutboxMapper extends BaseMapper<FileExportOutbox> {
|
||||
int sendFailed(@Param("id") long id, @Param("owner") String owner,
|
||||
@Param("backoffSeconds") int backoffSeconds, @Param("error") String error);
|
||||
|
||||
@Update("""
|
||||
UPDATE file_export_outbox SET next_retry_at=CURRENT_TIMESTAMP(3)
|
||||
WHERE task_id=#{taskId} AND dispatch_version=#{dispatchVersion}
|
||||
AND status='PENDING' AND next_retry_at>CURRENT_TIMESTAMP(3)
|
||||
""")
|
||||
int expeditePendingRetry(@Param("taskId") long taskId, @Param("dispatchVersion") int dispatchVersion);
|
||||
|
||||
@Update("UPDATE file_export_outbox SET status='PENDING',next_retry_at=CURRENT_TIMESTAMP(3),lease_owner=NULL,lease_until=NULL WHERE status='CLAIMED' AND lease_until<=CURRENT_TIMESTAMP(3) LIMIT 50")
|
||||
int releaseExpiredClaims();
|
||||
|
||||
|
||||
@@ -88,6 +88,13 @@ public interface FileExportTaskMapper extends BaseMapper<FileExportTask> {
|
||||
""")
|
||||
int manualRetry(@Param("id") long id);
|
||||
|
||||
@Update("""
|
||||
UPDATE file_export_task SET next_retry_at=CURRENT_TIMESTAMP(3)
|
||||
WHERE id=#{id} AND status='PENDING' AND dispatch_version=#{dispatchVersion}
|
||||
AND attempt_count>0 AND next_retry_at>CURRENT_TIMESTAMP(3)
|
||||
""")
|
||||
int expeditePendingRetry(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion);
|
||||
|
||||
@Update("UPDATE file_export_task SET status='FAILED',completed_at=CURRENT_TIMESTAMP(3),failure_message=#{reason},lease_owner=NULL,lease_until=NULL WHERE id=#{id} AND dispatch_version=#{dispatchVersion} AND status='PENDING'")
|
||||
int failQuarantined(@Param("id") long id, @Param("dispatchVersion") int dispatchVersion,
|
||||
@Param("reason") String reason);
|
||||
|
||||
@@ -22,6 +22,9 @@ public class ExportOutboxRepository {
|
||||
public boolean claim(long id, String owner, int leaseSeconds) { return mapper.claim(id, owner, leaseSeconds) == 1; }
|
||||
public boolean sent(long id, String owner) { return mapper.sent(id, owner) == 1; }
|
||||
public boolean sendFailed(long id, String owner, int seconds, String error) { return mapper.sendFailed(id, owner, seconds, error) == 1; }
|
||||
public boolean expeditePendingRetry(long taskId, int dispatchVersion) {
|
||||
return mapper.expeditePendingRetry(taskId, dispatchVersion) == 1;
|
||||
}
|
||||
public int releaseExpiredClaims() { return mapper.releaseExpiredClaims(); }
|
||||
public int redispatchSent(int staleSeconds, int graceSeconds) { return mapper.redispatchSent(staleSeconds, graceSeconds); }
|
||||
public boolean quarantine(String eventId, long taskId, int version, String reason) { return mapper.quarantine(eventId, taskId, version, reason) == 1; }
|
||||
|
||||
@@ -63,6 +63,9 @@ public class ExportTaskRepository {
|
||||
return mapper.fail(id, dispatchVersion, owner, message) == 1;
|
||||
}
|
||||
public boolean manualRetry(long id) { return mapper.manualRetry(id) == 1; }
|
||||
public boolean expeditePendingRetry(long id, int dispatchVersion) {
|
||||
return mapper.expeditePendingRetry(id, dispatchVersion) == 1;
|
||||
}
|
||||
public boolean failQuarantined(long id, int dispatchVersion, String reason) { return mapper.failQuarantined(id, dispatchVersion, reason) == 1; }
|
||||
public boolean recoverStale(long id, int dispatchVersion) { return mapper.recoverStale(id, dispatchVersion) == 1; }
|
||||
public boolean failStale(long id, int dispatchVersion) { return mapper.failStale(id, dispatchVersion) == 1; }
|
||||
|
||||
@@ -65,6 +65,14 @@ public class ExportStateService {
|
||||
return true;
|
||||
}
|
||||
@Transactional
|
||||
public boolean expeditePendingRetry(FileExportTask task) {
|
||||
if (!outbox.expeditePendingRetry(task.getId(), task.getDispatchVersion())) return false;
|
||||
if (!tasks.expeditePendingRetry(task.getId(), task.getDispatchVersion())) {
|
||||
throw new IllegalStateException("Unable to expedite pending export task");
|
||||
}
|
||||
return true;
|
||||
}
|
||||
@Transactional
|
||||
public void quarantine(String eventId, long taskId, int version, String reason) {
|
||||
FileExportTask current = tasks.find(taskId);
|
||||
if (current == null || current.getDispatchVersion() != version || !"PENDING".equals(current.getStatus())) return;
|
||||
|
||||
@@ -103,21 +103,27 @@ public class ExportTaskService {
|
||||
|
||||
public ExportTaskResponse retry(AuthenticatedUser user, long id) {
|
||||
FileExportTask task = ownedTask(user, id);
|
||||
if (!"FAILED".equals(task.getStatus())) {
|
||||
throw new BizException(ResultCode.CONFLICT, "只有失败的导出任务可以重试");
|
||||
}
|
||||
var definition = registry.require(task.getExportType());
|
||||
if (!definition.isEnabled()) throw new BizException(ResultCode.CONFLICT, "该导出类型已停用");
|
||||
String permission = definition.getRequiredPermission();
|
||||
if (!user.permissions().contains(permission)) {
|
||||
throw new BizException(ResultCode.FORBIDDEN, "缺少该导出类型的操作权限");
|
||||
}
|
||||
if ("PENDING".equals(task.getStatus()) && task.getAttemptCount() != null && task.getAttemptCount() > 0) {
|
||||
if (!stateService.expeditePendingRetry(task)) {
|
||||
throw new BizException(ResultCode.CONFLICT, "任务已进入执行队列,无需重复重试");
|
||||
}
|
||||
return response(repository.find(id));
|
||||
}
|
||||
if (!"FAILED".equals(task.getStatus())) {
|
||||
throw new BizException(ResultCode.CONFLICT, "只有等待重试或失败的导出任务可以重试");
|
||||
}
|
||||
if (!storage.delete(task.getObjectKey())) {
|
||||
throw new BizException(ResultCode.SERVICE_UNAVAILABLE, "上次导出文件清理失败,请稍后重试");
|
||||
}
|
||||
boolean retried = stateService.manualRetry(task);
|
||||
if (!retried) {
|
||||
throw new BizException(ResultCode.CONFLICT, "只有失败的导出任务可以重试");
|
||||
throw new BizException(ResultCode.CONFLICT, "只有等待重试或失败的导出任务可以重试");
|
||||
}
|
||||
return response(repository.find(id));
|
||||
}
|
||||
@@ -205,6 +211,7 @@ public class ExportTaskService {
|
||||
task.getFormat(), task.getTaskName(), task.getStatus(), task.getTotalRows(),
|
||||
task.getProcessedRows() == null ? 0 : task.getProcessedRows(), task.getFileName(),
|
||||
task.getContentType(), task.getSizeBytes(), task.getFailureMessage(),
|
||||
task.getAttemptCount() == null ? 0 : task.getAttemptCount(), task.getMaxAttempts(), task.getNextRetryAt(),
|
||||
String.valueOf(task.getRequestedBy()), task.getRequestedUsername(), task.getRequestedAt(),
|
||||
task.getStartedAt(), task.getCompletedAt(), task.getExpiresAt());
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user