fix(稳定性): 修任务锁超时/实例转发/扫描降噪,补 rustfs 结论日志
线上日志扫描发现的缺陷批量修复: - TaskFileJobService.resetStuckRunningJobsDetailed 去掉跨行长事务:两节点 @Scheduled 各取 200 行在同一事务里逐行 UPDATE,互等行锁导致 biz_task_file_job 每天上百次 Lock wait timeout。改逐行独立提交(本就有 CAS 条件,幂等) - TaskOwnerForwardService 连接类失败重试 3 次:两节点滚动重启窗口内的 Connection refused 会直接丢掉用户提交的结果(读超时不重放,避免重复提交) - AdminApiGuardFilter 把 401/被顶下线降为 debug(线上单节点一天近 3000 条噪声) - NoResourceFoundException 单独处理:如实返回 404,不再刷 ERROR 堆栈、不再伪装 200(每天 580 条,绝大多数是外部扫描 /.env、/credentials、aliyun.json 等) - RustfsObjectStorageService 补「重试后成功 / 重试耗尽」结论日志:一天上千条 retrying 却无结论,无法判断数据是否落盘 - 连接池 keep-alive 300s→30s,减少复用已被对端关闭的空闲连接 - SimilarAsin 提交结果遇到已删除任务时带 TASK_NOT_FOUND(40401),客户端据此 停止每 30 秒一轮的无谓重试 - 修 SimilarAsinResultRowDto 的 GBK 乱码 JsonAlias("浠锋牸" → "价格")
This commit is contained in:
@@ -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;
|
||||
}
|
||||
|
||||
+12
-2
@@ -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<ApiResponse<Void>> 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<Void> handleMethodArgumentNotValidException(MethodArgumentNotValidException ex) {
|
||||
String message = ex.getBindingResult().getFieldError() != null
|
||||
public ApiResponse<Void> handleMethodArgumentNotValidException(MethodArgumentNotValidException ex) { String message = ex.getBindingResult().getFieldError() != null
|
||||
? ex.getBindingResult().getFieldError().getDefaultMessage()
|
||||
: "参数校验失败";
|
||||
return ApiResponse.fail(message);
|
||||
|
||||
+55
@@ -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<String> 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 forwardWithConnectRetry(method, url, headers, body, ex);
|
||||
}
|
||||
|
||||
/**
|
||||
* 转发带连接级重试。
|
||||
*
|
||||
* <p>对端实例在部署窗口(原地换 JAR + 两节点滚动重启)内会有几秒的 Connection refused。
|
||||
* 连接都没建立起来说明请求没到达对端,此时重放是安全的;而读超时不重试——对端可能
|
||||
* 已经在处理,盲目重放会造成重复提交。线上由此丢过用户提交的结果。
|
||||
*/
|
||||
private ResponseEntity<byte[]> 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) {
|
||||
|
||||
@@ -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<Void> body = ex.getCode() == null
|
||||
? ApiResponse.fail(ex.getMessage())
|
||||
: ApiResponse.fail(ex.getCode(), 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)) {
|
||||
|
||||
@@ -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;
|
||||
|
||||
+9
@@ -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) {
|
||||
|
||||
+2
-1
@@ -277,7 +277,8 @@ public class SimilarAsinResultRowDto {
|
||||
@Schema(description = "阿里巴巴商品图片或商品 URL")
|
||||
private String url;
|
||||
|
||||
@JsonAlias({"price", "浠锋牸"})
|
||||
// "浠锋牸" 是 "价格" 被按 GBK 解读后的乱码,历史载荷里出现过,保留兼容
|
||||
@JsonAlias({"price", "价格", "浠锋牸"})
|
||||
@Schema(description = "阿里巴巴候选商品价格")
|
||||
private Object price;
|
||||
|
||||
|
||||
+4
-3
@@ -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={}",
|
||||
|
||||
+9
-2
@@ -298,12 +298,19 @@ public class TaskFileJobService {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Transactional
|
||||
public int resetStuckRunningJobs(int stuckMinutes, int limit) {
|
||||
return resetStuckRunningJobsDetailed(stuckMinutes, limit).resetCount();
|
||||
}
|
||||
|
||||
@Transactional
|
||||
/**
|
||||
* 卡住任务自愈扫描。
|
||||
*
|
||||
* <p>刻意**不加 @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<TaskFileJobEntity> jobs = taskFileJobMapper.selectList(new LambdaQueryWrapper<TaskFileJobEntity>()
|
||||
|
||||
Reference in New Issue
Block a user