diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/exception/BusinessCodes.java b/backend-java/src/main/java/com/nanri/aiimage/common/exception/BusinessCodes.java index ace43b4a..cd62b2c2 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/exception/BusinessCodes.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/exception/BusinessCodes.java @@ -25,4 +25,12 @@ public final class BusinessCodes { /** 任务归属其它实例，需转发。 */ public static final int TASK_OWNER_FORWARD = 40903; + + /** + * 提交结果的目标任务已不存在（通常是被删除）。 + * 语义：本次提交无意义，响应 success=false 且带该码，调用方应放弃而不是反复重试。 + * 与 {@link #TASK_ALREADY_FINISHED} 的区别：那个还留了任务记录（可幂等忽略）， + * 这个任务已经没了——如实报错，否则任务被误删时结果会被静默吞掉。 + */ + public static final int TASK_NOT_FOUND = 40401; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java b/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java index 030d125b..5ac6409e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java @@ -8,12 +8,14 @@ import jakarta.servlet.http.HttpServletRequest; import jakarta.validation.ConstraintViolationException; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.MethodArgumentNotValidException; import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.RestControllerAdvice; import org.springframework.web.context.request.async.AsyncRequestNotUsableException; import org.springframework.web.server.ResponseStatusException; +import org.springframework.web.servlet.resource.NoResourceFoundException; import java.io.IOException; @@ -85,6 +87,15 @@ public class GlobalExceptionHandler { : ApiResponse.fail(ex.getCode(), ex.getMessage()); } + @ExceptionHandler(NoResourceFoundException.class) + public ResponseEntity> handleNoResourceFoundException(NoResourceFoundException ex) { + // 静态资源 404。绝大部分是外部扫描器在探测 /.env、/credentials、aliyun.json、oss.json + // 这类云凭据文件（线上单节点一天 580 条）。此前落到 handleException 里，既刷 ERROR 堆栈， + // 又把探测响应伪装成 HTTP 200；这里降为 debug 并如实返回 404。 + log.debug("static resource not found: {}", ex.getMessage()); + return ResponseEntity.status(HttpStatus.NOT_FOUND).body(ApiResponse.fail("资源不存在")); + } + /** 未登录 / 登录态失效 / 被其他设备顶下线：按 401 语义的常态噪声，不占 WARN。 */ private boolean isRoutineAuthNoise(Integer code) { return Integer.valueOf(401).equals(code) @@ -92,8 +103,7 @@ public class GlobalExceptionHandler { } @ExceptionHandler(MethodArgumentNotValidException.class) - public ApiResponse handleMethodArgumentNotValidException(MethodArgumentNotValidException ex) { - String message = ex.getBindingResult().getFieldError() != null + public ApiResponse handleMethodArgumentNotValidException(MethodArgumentNotValidException ex) { String message = ex.getBindingResult().getFieldError() != null ? ex.getBindingResult().getFieldError().getDefaultMessage() : "参数校验失败"; return ApiResponse.fail(message); diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java b/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java index f16b2d5c..2dfc57d8 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java @@ -13,10 +13,12 @@ import org.springframework.http.ResponseEntity; import org.springframework.http.client.SimpleClientHttpRequestFactory; import org.springframework.stereotype.Service; import org.springframework.util.StreamUtils; +import org.springframework.web.client.ResourceAccessException; import org.springframework.web.client.RestClient; import org.springframework.web.util.ContentCachingRequestWrapper; import java.io.IOException; +import java.net.ConnectException; import java.util.Enumeration; import java.util.List; import java.util.Set; @@ -28,6 +30,11 @@ public class TaskOwnerForwardService { public static final String FORWARDED_HEADER = "X-AIIMAGE-Owner-Forwarded"; + /** 连接类失败的重试次数（含首次）。对端滚动重启时通常几秒内即可恢复。 */ + private static final int CONNECT_RETRY_TIMES = 3; + /** 第 n 次重试前的退避：1s、2s（总等待不超过 3s，不长时间占用请求线程）。 */ + private static final long CONNECT_RETRY_BACKOFF_MILLIS = 1000L; + private static final Set HOP_BY_HOP_HEADERS = Set.of( "connection", "keep-alive", @@ -58,12 +65,60 @@ public class TaskOwnerForwardService { HttpHeaders headers = copyForwardHeaders(request, ex.getCurrentInstanceId()); log.info("[instance-routing] forward {} taskId={} operation={} owner={} current={} url={}", method, ex.getTaskId(), ex.getOperation(), ex.getOwnerInstanceId(), ex.getCurrentInstanceId(), url); - return restClient().method(method) - .uri(url) - .headers(target -> target.addAll(headers)) - .body(body) - .retrieve() - .toEntity(byte[].class); + return forwardWithConnectRetry(method, url, headers, body, ex); + } + + /** + * 转发带连接级重试。 + * + *

对端实例在部署窗口（原地换 JAR + 两节点滚动重启）内会有几秒的 Connection refused。 + * 连接都没建立起来说明请求没到达对端，此时重放是安全的；而读超时不重试——对端可能 + * 已经在处理，盲目重放会造成重复提交。线上由此丢过用户提交的结果。 + */ + private ResponseEntity forwardWithConnectRetry(HttpMethod method, String url, + HttpHeaders headers, byte[] body, + TaskOwnerMismatchException ex) { + RuntimeException lastError = null; + for (int attempt = 1; attempt <= CONNECT_RETRY_TIMES; attempt++) { + try { + return restClient().method(method) + .uri(url) + .headers(target -> target.addAll(headers)) + .body(body) + .retrieve() + .toEntity(byte[].class); + } catch (ResourceAccessException accessError) { + if (!isConnectFailure(accessError)) { + throw accessError; + } + lastError = accessError; + log.warn("[instance-routing] 转发连接失败，第 {}/{} 次 url={} taskId={} 原因={}", + attempt, CONNECT_RETRY_TIMES, url, ex.getTaskId(), accessError.getMessage()); + if (attempt < CONNECT_RETRY_TIMES) { + sleepQuietly(CONNECT_RETRY_BACKOFF_MILLIS * attempt); + } + } + } + throw lastError; + } + + private static boolean isConnectFailure(Throwable error) { + Throwable cursor = error; + while (cursor != null) { + if (cursor instanceof ConnectException) { + return true; + } + cursor = cursor.getCause(); + } + return false; + } + + private static void sleepQuietly(long millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + } } private String resolveUrl(TaskOwnerMismatchException ex, String path) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/AdminApiGuardFilter.java b/backend-java/src/main/java/com/nanri/aiimage/config/AdminApiGuardFilter.java index 3bd3e271..f8fb6a30 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/AdminApiGuardFilter.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/AdminApiGuardFilter.java @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.api.ApiResponse; import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.common.security.AdminAuthSupport; +import com.nanri.aiimage.common.security.DeviceSessionPolicy; import jakarta.servlet.FilterChain; import jakarta.servlet.ServletException; import jakarta.servlet.http.HttpServletRequest; @@ -153,7 +154,14 @@ public class AdminApiGuardFilter extends OncePerRequestFilter { ApiResponse body = ex.getCode() == null ? ApiResponse.fail(ex.getMessage()) : ApiResponse.fail(ex.getCode(), ex.getMessage()); - log.warn("[admin-guard] {} {} rejected: {}", request.getMethod(), request.getRequestURI(), ex.getMessage()); + // 401（登录已过期）与被顶下线是前端定时轮询（/api/notifications/summary、/api/user-secrets 等） + // 的常态：线上单节点一天近 3000 条，会把真实业务错误淹没。与 GlobalExceptionHandler + // 的 isRoutineAuthNoise 同一口径降为 debug。 + if (isRoutineAuthNoise(ex.getCode())) { + log.debug("[admin-guard] {} {} rejected: {}", request.getMethod(), request.getRequestURI(), ex.getMessage()); + } else { + log.warn("[admin-guard] {} {} rejected: {}", request.getMethod(), request.getRequestURI(), ex.getMessage()); + } response.setStatus(HttpServletResponse.SC_OK); response.setContentType("application/json;charset=UTF-8"); response.getWriter().write(objectMapper.writeValueAsString(body)); @@ -168,6 +176,12 @@ public class AdminApiGuardFilter extends OncePerRequestFilter { chain.doFilter(request, response); } + /** 登录态过期 / 被其他设备顶下线：前端轮询的常态噪声，不占 WARN。 */ + private static boolean isRoutineAuthNoise(Integer code) { + return Integer.valueOf(401).equals(code) + || Integer.valueOf(DeviceSessionPolicy.CODE_KICKED).equals(code); + } + /** 命中受保护前缀（/api/admin、/debug、自助接口、用户态工具前缀及其子路径）才进入鉴权，其余请求直接放行。 */ private boolean isGuarded(String uri) { if (matchesPrefix(uri, ADMIN_API_PREFIX) || matchesPrefix(uri, DEBUG_PREFIX)) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java index 7c6fc927..aab05e41 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java @@ -38,7 +38,12 @@ public class TransientStorageProperties { private int dispatcherMaxRequests = 56; private int dispatcherMaxRequestsPerHost = 56; private int connectionPoolMaxIdle = 5; - private long connectionPoolKeepAliveMillis = 300000; + /** + * 空闲连接在池里的保留时长。默认 5 分钟（OkHttp 原值）会让客户端复用"已被 RustFS + * 或中间设备关掉的空闲连接"，表现为 unexpected end of stream —— 线上每天上千次， + * 全靠重试兜底。对象存储访问是突发型，连接复用率本就低，缩短保活几乎没有代价。 + */ + private long connectionPoolKeepAliveMillis = 30000; private long warnPayloadBytes = 5L * 1024 * 1024; private long maxPayloadBytes = 50L * 1024 * 1024; private long maxStoredPayloadBytes = 50L * 1024 * 1024; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java index ba0743b1..dc7f9145 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java @@ -238,6 +238,12 @@ public class RustfsObjectStorageService { resetFailureWindow(operation); } recordOperation(operation, "success", elapsedNanos(startedAt)); + if (attempt > 1) { + // 重试后成功必须留 INFO 结论：线上一天上千条 "operation failed, retrying" + // 却没有任何结论日志，无法判断这些上传最后到底落盘了没有。 + log.info("[rustfs] 重试后成功 operation={} objectKey={} attempt={}/{} durationMs={} bytes={}", + operation, objectKey, attempt, maxRetries, elapsedMillis(startedAt), bytes); + } log.debug("[rustfs] operation success operation={} objectKey={} attempt={}/{} durationMs={} bytes={}", operation, objectKey, attempt, maxRetries, elapsedMillis(startedAt), bytes); return result; @@ -262,6 +268,9 @@ public class RustfsObjectStorageService { operation, objectKey, deadlineNanos); } } + // 重试耗尽：留 ERROR 结论 + 最后一次错误，否则只看到一串 retrying，无从判断是否真丢数据 + log.error("[rustfs] 重试耗尽，最终失败 operation={} objectKey={} maxRetries={} 最后一次错误={}", + operation, objectKey, maxRetries, last == null ? "无" : last.getMessage(), last); throw new IllegalStateException("failed to " + operation + " payload in transient storage", last); } finally { if (totalAcquired) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java index 008d05b1..a395b046 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java @@ -277,7 +277,8 @@ public class SimilarAsinResultRowDto { @Schema(description = "阿里巴巴商品图片或商品 URL") private String url; - @JsonAlias({"price", "浠锋牸"}) + // "浠锋牸" 是 "价格" 被按 GBK 解读后的乱码，历史载荷里出现过，保留兼容 + @JsonAlias({"price", "价格", "浠锋牸"}) @Schema(description = "阿里巴巴候选商品价格") private Object price; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinPipelineSupport.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinPipelineSupport.java index 3f46952b..c5fd8e27 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinPipelineSupport.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinPipelineSupport.java @@ -15,6 +15,7 @@ import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.exception.BusinessCodes; import com.nanri.aiimage.common.service.DistributedJobLockService; import com.nanri.aiimage.config.SimilarAsinProperties; import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto; @@ -468,7 +469,7 @@ public class SimilarAsinPipelineSupport { public PreparedSubmittedChunk prepareSubmittedChunk(Long taskId, SimilarAsinSubmitResultRequest request) { FileTaskEntity task = fileTaskMapper.selectById(taskId); if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { - throw new BusinessException("任务不存在"); + throw new BusinessException(BusinessCodes.TASK_NOT_FOUND, "任务不存在"); } if (!STATUS_RUNNING.equals(task.getStatus())) { throw new BusinessException("任务不是运行中状态"); @@ -509,7 +510,7 @@ public class SimilarAsinPipelineSupport { Long taskId = prepared.taskId(); FileTaskEntity task = fileTaskMapper.selectById(taskId); if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { - throw new BusinessException("任务不存在"); + throw new BusinessException(BusinessCodes.TASK_NOT_FOUND, "任务不存在"); } if (!STATUS_RUNNING.equals(task.getStatus())) { throw new BusinessException("任务不是运行中状态"); @@ -561,7 +562,7 @@ public class SimilarAsinPipelineSupport { public FinalizeTaskResult completeSubmittedChunk(SubmitContext context) { FileTaskEntity task = fileTaskMapper.selectById(context.task().getId()); if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { - throw new BusinessException("任务不存在"); + throw new BusinessException(BusinessCodes.TASK_NOT_FOUND, "任务不存在"); } if (!STATUS_RUNNING.equals(task.getStatus())) { log.info("[similar-asin] skip completion because task already finalized taskId={} status={}", diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java index 1df2cdc0..8066b283 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java @@ -298,12 +298,19 @@ public class TaskFileJobService { return false; } - @Transactional public int resetStuckRunningJobs(int stuckMinutes, int limit) { return resetStuckRunningJobsDetailed(stuckMinutes, limit).resetCount(); } - @Transactional + /** + * 卡住任务自愈扫描。 + * + *

刻意**不加 @Transactional**：原实现在一个事务里 SELECT 最多 200 行再逐行 UPDATE， + * 而两台实例各跑一份 @Scheduled，双方拿到同一批 RUNNING 行后互相等行锁，直接造成线上 + * 每天上百次 "Lock wait timeout exceeded"（biz_task_file_job）。 + * 这里的每行更新本就带 status + updatedAt 的 CAS 条件（幂等），逐行独立提交更安全： + * 锁即时释放，CAS 不匹配的一方返回 0 行即可，也不会因中途异常回滚掉已修好的行。 + */ public StuckJobResetResult resetStuckRunningJobsDetailed(int stuckMinutes, int limit) { LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(5, stuckMinutes)); List jobs = taskFileJobMapper.selectList(new LambdaQueryWrapper()