From cf77e1976524586cdee13971a7e9db10302721cd Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Thu, 3 Sep 2026 10:40:21 +0800 Subject: [PATCH] feat(file): expedite pending export retries --- .../java/com/baiyue/file/dto/ExportTaskDtos.java | 1 + .../file/mapper/FileExportOutboxMapper.java | 7 +++++++ .../baiyue/file/mapper/FileExportTaskMapper.java | 7 +++++++ .../file/repository/ExportOutboxRepository.java | 3 +++ .../file/repository/ExportTaskRepository.java | 3 +++ .../baiyue/file/service/ExportStateService.java | 8 ++++++++ .../baiyue/file/service/ExportTaskService.java | 15 +++++++++++---- 7 files changed, 40 insertions(+), 4 deletions(-) diff --git a/file/src/main/java/com/baiyue/file/dto/ExportTaskDtos.java b/file/src/main/java/com/baiyue/file/dto/ExportTaskDtos.java index d1079c543..beb48161b 100644 --- a/file/src/main/java/com/baiyue/file/dto/ExportTaskDtos.java +++ b/file/src/main/java/com/baiyue/file/dto/ExportTaskDtos.java @@ -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) { } } diff --git a/file/src/main/java/com/baiyue/file/mapper/FileExportOutboxMapper.java b/file/src/main/java/com/baiyue/file/mapper/FileExportOutboxMapper.java index 5ed9e1503..eaffd1079 100644 --- a/file/src/main/java/com/baiyue/file/mapper/FileExportOutboxMapper.java +++ b/file/src/main/java/com/baiyue/file/mapper/FileExportOutboxMapper.java @@ -33,6 +33,13 @@ public interface FileExportOutboxMapper extends BaseMapper { 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(); diff --git a/file/src/main/java/com/baiyue/file/mapper/FileExportTaskMapper.java b/file/src/main/java/com/baiyue/file/mapper/FileExportTaskMapper.java index c1568adf4..25447db04 100644 --- a/file/src/main/java/com/baiyue/file/mapper/FileExportTaskMapper.java +++ b/file/src/main/java/com/baiyue/file/mapper/FileExportTaskMapper.java @@ -88,6 +88,13 @@ public interface FileExportTaskMapper extends BaseMapper { """) 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); diff --git a/file/src/main/java/com/baiyue/file/repository/ExportOutboxRepository.java b/file/src/main/java/com/baiyue/file/repository/ExportOutboxRepository.java index 9eac22349..81fa1cc52 100644 --- a/file/src/main/java/com/baiyue/file/repository/ExportOutboxRepository.java +++ b/file/src/main/java/com/baiyue/file/repository/ExportOutboxRepository.java @@ -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; } diff --git a/file/src/main/java/com/baiyue/file/repository/ExportTaskRepository.java b/file/src/main/java/com/baiyue/file/repository/ExportTaskRepository.java index be4797a05..7fe987e8f 100644 --- a/file/src/main/java/com/baiyue/file/repository/ExportTaskRepository.java +++ b/file/src/main/java/com/baiyue/file/repository/ExportTaskRepository.java @@ -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; } diff --git a/file/src/main/java/com/baiyue/file/service/ExportStateService.java b/file/src/main/java/com/baiyue/file/service/ExportStateService.java index d03ed3856..62bfb5ae5 100644 --- a/file/src/main/java/com/baiyue/file/service/ExportStateService.java +++ b/file/src/main/java/com/baiyue/file/service/ExportStateService.java @@ -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; diff --git a/file/src/main/java/com/baiyue/file/service/ExportTaskService.java b/file/src/main/java/com/baiyue/file/service/ExportTaskService.java index 22071a29a..a25401d95 100644 --- a/file/src/main/java/com/baiyue/file/service/ExportTaskService.java +++ b/file/src/main/java/com/baiyue/file/service/ExportTaskService.java @@ -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()); }