Compare commits

...

5 Commits

13 changed files with 382 additions and 122 deletions
@@ -44,8 +44,14 @@ public class BrandCheckClient {
this.externalCallMetrics = externalCallMetrics; this.externalCallMetrics = externalCallMetrics;
} }
private final ExecutorService checkExecutor = Executors.newFixedThreadPool( // 有界队列 + CallerRuns:大批量(几千品牌)检查时队列洪峰不无界堆内存,
BRAND_CHECK_CONCURRENCY, namedThreadFactory("brand-check")); // 队列打满时提交线程自己执行一个检查任务,上游天然限速。
private final ExecutorService checkExecutor = new java.util.concurrent.ThreadPoolExecutor(
BRAND_CHECK_CONCURRENCY, BRAND_CHECK_CONCURRENCY,
60L, java.util.concurrent.TimeUnit.SECONDS,
new java.util.concurrent.LinkedBlockingQueue<>(1000),
namedThreadFactory("brand-check"),
new java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy());
@PreDestroy @PreDestroy
void shutdownCheckExecutor() { void shutdownCheckExecutor() {
@@ -58,6 +58,7 @@ import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import com.nanri.aiimage.modules.task.service.TaskFileJobService; import com.nanri.aiimage.modules.task.service.TaskFileJobService;
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.usermodel.Cell; import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.ss.usermodel.DataFormatter; import org.apache.poi.ss.usermodel.DataFormatter;
@@ -147,6 +148,7 @@ public class CollectDataService {
private final OssStorageService ossStorageService; private final OssStorageService ossStorageService;
private final ObjectMapper objectMapper; private final ObjectMapper objectMapper;
private final TransactionTemplate transactionTemplate; private final TransactionTemplate transactionTemplate;
private final DistributedJobLockService distributedJobLockService;
/** ASIN 去重 + 无效品牌批量集合查询器:两段式查询合并为一次往返,语义与旧实现等价。 */ /** ASIN 去重 + 无效品牌批量集合查询器:两段式查询合并为一次往返,语义与旧实现等价。 */
private final CollectDataBatchQuery collectDataBatchQuery; private final CollectDataBatchQuery collectDataBatchQuery;
@@ -483,6 +485,14 @@ public class CollectDataService {
@Scheduled(cron = "${aiimage.collect-data.stale-check-cron:*/30 * * * * *}") @Scheduled(cron = "${aiimage.collect-data.stale-check-cron:*/30 * * * * *}")
public void finalizeStaleTasks() { public void finalizeStaleTasks() {
// 双节点互斥:Redis 不可用时 fail-open(仅记日志),兜底靠下方 per-task DB 锁,不会双写终态
DistributedJobLockService.LockHandle jobLock =
distributedJobLockService.tryLock("collect-data:stale-check", java.time.Duration.ofMinutes(2));
if (jobLock == null) {
log.info("[collect-data] stale-check skipped, another instance holds the distributed lock");
return;
}
try (jobLock) {
long timeoutMinutes = Math.max(1L, staleTimeoutMinutes); long timeoutMinutes = Math.max(1L, staleTimeoutMinutes);
LocalDateTime threshold = LocalDateTime.now().minusMinutes(timeoutMinutes); LocalDateTime threshold = LocalDateTime.now().minusMinutes(timeoutMinutes);
List<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>() List<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
@@ -495,6 +505,7 @@ public class CollectDataService {
finalizeStaleTask(task.getId(), threshold, timeoutMinutes); finalizeStaleTask(task.getId(), threshold, timeoutMinutes);
} }
} }
}
private void finalizeStaleTask(Long taskId, LocalDateTime threshold, long timeoutMinutes) { private void finalizeStaleTask(Long taskId, LocalDateTime threshold, long timeoutMinutes) {
try (TaskDistributedLockService.LockHandle lock = try (TaskDistributedLockService.LockHandle lock =
@@ -1201,36 +1212,6 @@ public class CollectDataService {
return safe.isBlank() ? "collect-data" : safe; return safe.isBlank() ? "collect-data" : safe;
} }
private void deleteTransientTaskPayloads(List<TaskChunkEntity> chunks, List<TaskResultItemEntity> items) {
if (chunks != null) {
Set<String> deleted = new HashSet<>();
for (TaskChunkEntity chunk : chunks) {
// 多 chunk 共享同一对象(deterministic key 残留场景)按值去重只删一次。
if (chunk.getPayloadJson() != null && deleted.add(chunk.getPayloadJson())) {
transientPayloadStorageService.deletePayloadIfPresent(chunk.getPayloadJson());
}
}
}
deleteResultItemPayloads(items);
}
/** 删除结果明细 payload:chunk 级引用按对象去重后各删一次,旧格式逐行删。 */
private void deleteResultItemPayloads(List<TaskResultItemEntity> items) {
if (items == null) {
return;
}
Set<String> deletedPointers = new HashSet<>();
for (TaskResultItemEntity item : items) {
CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(item.getPayloadJson());
if (ref != null) {
if (deletedPointers.add(ref.pointer())) {
transientPayloadStorageService.deletePayloadIfPresent(ref.pointer());
}
} else {
transientPayloadStorageService.deletePayloadIfPresent(item.getPayloadJson());
}
}
}
private static class CollectDataStats { private static class CollectDataStats {
private int totalRows; private int totalRows;
@@ -1252,9 +1233,38 @@ public class CollectDataService {
private List<CollectDataSummaryRowDto> summaries = new ArrayList<>(); private List<CollectDataSummaryRowDto> summaries = new ArrayList<>();
} }
@Transactional // 不用 @Transactional(同类自调用代理失效):用 TransactionTemplate 显式包 DB 段,
// 事务提交后再做远程删除(慢 IO),避免大任务删除时长事务占用连接池。
public void deleteTask(Long taskId, Long userId) { public void deleteTask(Long taskId, Long userId) {
List<String> chunkPayloadPointers = new ArrayList<>();
List<String> resultPayloadJsons = new ArrayList<>();
transactionTemplate.executeWithoutResult(status ->
deleteTaskRows(taskId, userId, chunkPayloadPointers, resultPayloadJsons));
// 事务已提交,这里执行远程删除
Set<String> deleted = new HashSet<>();
for (String pointer : chunkPayloadPointers) {
if (deleted.add(pointer)) {
transientPayloadStorageService.deletePayloadIfPresent(pointer);
}
}
deletePayloadJsonsAndPointers(resultPayloadJsons, deleted);
}
private void deleteTaskRows(Long taskId, Long userId, List<String> chunkPayloadPointers, List<String> resultPayloadJsons) {
FileTaskEntity task = requireTask(taskId, userId); FileTaskEntity task = requireTask(taskId, userId);
taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
.select(TaskChunkEntity::getPayloadJson)
.eq(TaskChunkEntity::getTaskId, task.getId())
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)).stream()
.map(TaskChunkEntity::getPayloadJson)
.filter(Objects::nonNull)
.forEach(chunkPayloadPointers::add);
taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.select(TaskResultItemEntity::getPayloadJson)
.eq(TaskResultItemEntity::getTaskId, task.getId())
.in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)).stream()
.map(TaskResultItemEntity::getPayloadJson)
.forEach(resultPayloadJsons::add);
collectDataItemMapper.delete(new LambdaQueryWrapper<CollectDataItemEntity>() collectDataItemMapper.delete(new LambdaQueryWrapper<CollectDataItemEntity>()
.eq(CollectDataItemEntity::getTaskId, task.getId())); .eq(CollectDataItemEntity::getTaskId, task.getId()));
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>() fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
@@ -1271,24 +1281,32 @@ public class CollectDataService {
taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>() taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getTaskId, task.getId()) .eq(TaskResultItemEntity::getTaskId, task.getId())
.in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)); .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE));
deleteTransientTaskPayloads(
taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
.select(TaskChunkEntity::getPayloadJson)
.eq(TaskChunkEntity::getTaskId, task.getId())
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)),
taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.select(TaskResultItemEntity::getPayloadJson)
.eq(TaskResultItemEntity::getTaskId, task.getId())
.in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)));
taskFileJobService.deleteTaskJobs(task.getId(), MODULE_TYPE); taskFileJobService.deleteTaskJobs(task.getId(), MODULE_TYPE);
fileTaskMapper.deleteById(task.getId()); fileTaskMapper.deleteById(task.getId());
} }
private void deletePayloadJsonsAndPointers(List<String> payloadJsons, Set<String> deletedPointers) {
for (String payloadJson : payloadJsons) {
if (payloadJson == null) {
continue;
}
CollectDataResultDetailCodec.ChunkRef ref = resultDetailCodec.parseRef(payloadJson);
if (ref != null) {
if (deletedPointers.add(ref.pointer())) {
transientPayloadStorageService.deletePayloadIfPresent(ref.pointer());
}
} else {
transientPayloadStorageService.deletePayloadIfPresent(payloadJson);
}
}
}
public void deleteHistory(Long resultId, Long userId) { public void deleteHistory(Long resultId, Long userId) {
FileResultEntity row = fileResultMapper.selectById(resultId); FileResultEntity row = fileResultMapper.selectById(resultId);
if (row == null || !MODULE_TYPE.equals(row.getModuleType()) || !Objects.equals(userId, row.getUserId())) { if (row == null || !MODULE_TYPE.equals(row.getModuleType()) || !Objects.equals(userId, row.getUserId())) {
throw new BusinessException("记录不存在"); throw new BusinessException("记录不存在");
} }
List<String> resultPayloadJsons = transactionTemplate.execute(status -> {
List<TaskResultItemEntity> resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>() List<TaskResultItemEntity> resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.select(TaskResultItemEntity::getPayloadJson) .select(TaskResultItemEntity::getPayloadJson)
.eq(TaskResultItemEntity::getTaskId, row.getTaskId()) .eq(TaskResultItemEntity::getTaskId, row.getTaskId())
@@ -1298,10 +1316,12 @@ public class CollectDataService {
.eq(TaskResultItemEntity::getTaskId, row.getTaskId()) .eq(TaskResultItemEntity::getTaskId, row.getTaskId())
.in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)
.eq(TaskResultItemEntity::getResultId, row.getId())); .eq(TaskResultItemEntity::getResultId, row.getId()));
// 与 deleteTask 一致:先删 DB 行再物理删对象,保证行删除与对象删除一致。
deleteResultItemPayloads(resultItems);
taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, row.getId()); taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, row.getId());
fileResultMapper.deleteById(resultId); fileResultMapper.deleteById(resultId);
return resultItems.stream().map(TaskResultItemEntity::getPayloadJson).toList();
});
// 事务已提交,再物理删对象,保证行删除与对象删除一致。
deletePayloadJsonsAndPointers(resultPayloadJsons, new HashSet<>());
} }
public CollectDataCountryPreferenceVo getCountryPreference(Long userId) { public CollectDataCountryPreferenceVo getCountryPreference(Long userId) {
@@ -41,6 +41,8 @@ public class OssStorageService {
private static final String DIGITAL_HUMAN_PREFIX = "digital-human/versions/"; private static final String DIGITAL_HUMAN_PREFIX = "digital-human/versions/";
private static final String SOFTWARE_VERSION_PREFIX = "nanri-image/versions/"; private static final String SOFTWARE_VERSION_PREFIX = "nanri-image/versions/";
private static final String LEGACY_MINIO_ENDPOINT = "http://47.110.241.161:9000"; private static final String LEGACY_MINIO_ENDPOINT = "http://47.110.241.161:9000";
/** readObjectBytes 兜底读取上限(20MB):防止大对象一次性读入内存导致 OOM。 */
private static final long DEFAULT_READ_MAX_BYTES = 20L * 1024 * 1024;
private final OssProperties ossProperties; private final OssProperties ossProperties;
private final MinioClient minioClient; private final MinioClient minioClient;
@@ -248,13 +250,32 @@ public class OssStorageService {
} }
public byte[] readObjectBytes(String bucket, String objectKey) { public byte[] readObjectBytes(String bucket, String objectKey) {
// 防御性上限:防止被误用到大对象时一次性读入内存导致 OOM(当前仅模板下载场景,模板为 xlsx KB 级)
return readObjectBytes(bucket, objectKey, DEFAULT_READ_MAX_BYTES);
}
public byte[] readObjectBytes(String bucket, String objectKey, long maxBytes) {
String normalizedBucket = requireStorageName(bucket, "bucket"); String normalizedBucket = requireStorageName(bucket, "bucket");
String normalizedObjectKey = requireStorageName(objectKey, "objectKey"); String normalizedObjectKey = requireStorageName(objectKey, "objectKey");
try (var stream = buildClient().getObject(GetObjectArgs.builder() try (var stream = buildClient().getObject(GetObjectArgs.builder()
.bucket(normalizedBucket) .bucket(normalizedBucket)
.object(normalizedObjectKey) .object(normalizedObjectKey)
.build())) { .build())) {
return stream.readAllBytes(); byte[] buffer = new byte[1024 * 1024];
ByteArrayOutputStream out = new ByteArrayOutputStream();
long total = 0;
int read;
while ((read = stream.read(buffer)) != -1) {
total += read;
if (total > maxBytes) {
throw new IllegalArgumentException("对象过大,超过读取上限 " + maxBytes + " 字节: "
+ normalizedBucket + "/" + normalizedObjectKey);
}
out.write(buffer, 0, read);
}
return out.toByteArray();
} catch (IllegalArgumentException ex) {
throw ex;
} catch (Exception ex) { } catch (Exception ex) {
throw storageFailure("read", normalizedBucket + "/" + normalizedObjectKey, ex); throw storageFailure("read", normalizedBucket + "/" + normalizedObjectKey, ex);
} }
@@ -16,6 +16,7 @@ import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoWorkflowResultRe
import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoWorkflowRunRequest; import com.nanri.aiimage.modules.imagevideo.model.dto.ImageVideoWorkflowRunRequest;
import com.nanri.aiimage.modules.imagevideo.model.entity.ImageVideoAsyncTaskEntity; import com.nanri.aiimage.modules.imagevideo.model.entity.ImageVideoAsyncTaskEntity;
import com.nanri.aiimage.modules.imagevideo.model.vo.ImageVideoAsyncTaskVo; import com.nanri.aiimage.modules.imagevideo.model.vo.ImageVideoAsyncTaskVo;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Qualifier;
@@ -66,6 +67,7 @@ public class ImageVideoAsyncTaskService {
private final ObjectMapper objectMapper; private final ObjectMapper objectMapper;
private final TaskExecutor taskQueueExecutor; private final TaskExecutor taskQueueExecutor;
private final InstanceMetadata instanceMetadata; private final InstanceMetadata instanceMetadata;
private final com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService;
public ImageVideoAsyncTaskService( public ImageVideoAsyncTaskService(
ImageVideoAsyncTaskMapper taskMapper, ImageVideoAsyncTaskMapper taskMapper,
@@ -74,7 +76,8 @@ public class ImageVideoAsyncTaskService {
ImageVideoArchiveService archiveService, ImageVideoArchiveService archiveService,
ObjectMapper objectMapper, ObjectMapper objectMapper,
@Qualifier("taskQueueExecutor") TaskExecutor taskQueueExecutor, @Qualifier("taskQueueExecutor") TaskExecutor taskQueueExecutor,
InstanceMetadata instanceMetadata) { InstanceMetadata instanceMetadata,
com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService) {
this.taskMapper = taskMapper; this.taskMapper = taskMapper;
this.cozeService = cozeService; this.cozeService = cozeService;
this.workflowConfigService = workflowConfigService; this.workflowConfigService = workflowConfigService;
@@ -82,6 +85,7 @@ public class ImageVideoAsyncTaskService {
this.objectMapper = objectMapper; this.objectMapper = objectMapper;
this.taskQueueExecutor = taskQueueExecutor; this.taskQueueExecutor = taskQueueExecutor;
this.instanceMetadata = instanceMetadata; this.instanceMetadata = instanceMetadata;
this.distributedJobLockService = distributedJobLockService;
} }
public ImageVideoAsyncTaskVo submitDouyinCopy(DouyinCopyRequest request) { public ImageVideoAsyncTaskVo submitDouyinCopy(DouyinCopyRequest request) {
@@ -130,6 +134,13 @@ public class ImageVideoAsyncTaskService {
@Scheduled(fixedDelayString = "${aiimage.image-video.async-task-dispatch-delay-ms:1000}") @Scheduled(fixedDelayString = "${aiimage.image-video.async-task-dispatch-delay-ms:1000}")
public void dispatchPendingTasks() { public void dispatchPendingTasks() {
// 双节点互斥:只影响扫描入队动作(CAS claim 已兜底不重复执行),省掉另一节点的每秒空扫
DistributedJobLockService.LockHandle jobLock =
distributedJobLockService.tryLock("image-video:dispatch", java.time.Duration.ofSeconds(5));
if (jobLock == null) {
return;
}
try (jobLock) {
List<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>() List<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
.eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.PENDING.name()) .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.PENDING.name())
.and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId()) .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId())
@@ -139,9 +150,16 @@ public class ImageVideoAsyncTaskService {
.last("LIMIT " + DISPATCH_BATCH_SIZE)); .last("LIMIT " + DISPATCH_BATCH_SIZE));
tasks.forEach(task -> taskQueueExecutor.execute(() -> executeTask(task.getId()))); tasks.forEach(task -> taskQueueExecutor.execute(() -> executeTask(task.getId())));
} }
}
@Scheduled(fixedDelayString = "${aiimage.image-video.async-task-poll-delay-ms:5000}") @Scheduled(fixedDelayString = "${aiimage.image-video.async-task-poll-delay-ms:5000}")
public void pollWaitingTasks() { public void pollWaitingTasks() {
DistributedJobLockService.LockHandle jobLock =
distributedJobLockService.tryLock("image-video:poll", java.time.Duration.ofSeconds(30));
if (jobLock == null) {
return;
}
try (jobLock) {
List<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>() List<ImageVideoAsyncTaskEntity> tasks = taskMapper.selectList(new LambdaQueryWrapper<ImageVideoAsyncTaskEntity>()
.eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.WAITING.name()) .eq(ImageVideoAsyncTaskEntity::getStatus, TaskStatus.WAITING.name())
.and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId()) .and(q -> q.eq(ImageVideoAsyncTaskEntity::getOwnerInstanceId, currentInstanceId())
@@ -151,6 +169,7 @@ public class ImageVideoAsyncTaskService {
.last("LIMIT " + POLL_BATCH_SIZE)); .last("LIMIT " + POLL_BATCH_SIZE));
tasks.forEach(task -> taskQueueExecutor.execute(() -> pollTask(task.getId()))); tasks.forEach(task -> taskQueueExecutor.execute(() -> pollTask(task.getId())));
} }
}
@EventListener(ApplicationReadyEvent.class) @EventListener(ApplicationReadyEvent.class)
public void recoverInterruptedPollingTasks() { public void recoverInterruptedPollingTasks() {
@@ -50,6 +50,7 @@ import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopIndexService;
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import org.springframework.beans.factory.annotation.Value; import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException; import org.springframework.dao.DuplicateKeyException;
import org.springframework.scheduling.annotation.Scheduled; import org.springframework.scheduling.annotation.Scheduled;
@@ -108,6 +109,7 @@ public class PublishTaskService {
private final OssStorageService ossStorageService; private final OssStorageService ossStorageService;
private final ObjectMapper objectMapper; private final ObjectMapper objectMapper;
private final TransactionTemplate transactionTemplate; private final TransactionTemplate transactionTemplate;
private final DistributedJobLockService distributedJobLockService;
private final InstanceMetadata instanceMetadata; private final InstanceMetadata instanceMetadata;
private final TaskProgressLightAssembler taskProgressLightAssembler; private final TaskProgressLightAssembler taskProgressLightAssembler;
@@ -420,6 +422,15 @@ public class PublishTaskService {
@Scheduled(fixedDelayString = "${aiimage.publish.stale-scan-delay-ms:60000}") @Scheduled(fixedDelayString = "${aiimage.publish.stale-scan-delay-ms:60000}")
public void failStaleTasks() { public void failStaleTasks() {
// 双节点互斥:避免 ownerInstanceId 为空的 stale 行被两节点同时扫出双写错误信息;
// Redis 不可用时 fail-open,兜底靠下方 per-task 锁
DistributedJobLockService.LockHandle jobLock =
distributedJobLockService.tryLock("publish:stale-check", java.time.Duration.ofMinutes(2));
if (jobLock == null) {
log.info("[publish] stale-check skipped, another instance holds the distributed lock");
return;
}
try (jobLock) {
LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(5, staleTimeoutMinutes)); LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(5, staleTimeoutMinutes));
List<FileTaskEntity> staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>() List<FileTaskEntity> staleTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE) .eq(FileTaskEntity::getModuleType, MODULE_TYPE)
@@ -449,6 +460,7 @@ public class PublishTaskService {
} }
} }
} }
}
public void processResultFileJob(TaskFileJobEntity job) { public void processResultFileJob(TaskFileJobEntity job) {
if (job == null || job.getTaskId() == null || job.getResultId() == null) { if (job == null || job.getTaskId() == null || job.getResultId() == null) {
@@ -33,4 +33,17 @@ public class TaskHeartbeatController {
@Valid @RequestBody(required = false) TaskHeartbeatRequest request) { @Valid @RequestBody(required = false) TaskHeartbeatRequest request) {
return ApiResponse.success(taskHeartbeatService.heartbeat(taskId, request)); return ApiResponse.success(taskHeartbeatService.heartbeat(taskId, request));
} }
@PostMapping("/{taskId}/interrupted")
@Operation(
summary = "上报客户端异常中断",
description = "客户端重启后发现自己上次进程崩溃时调用:将仍在 RUNNING 的任务立即标为终态"
+ "file_task→FAILEDbrand_crawl_tasks→cancelled),替代最长 30 分钟的 stale 兜底。幂等,非 RUNNING 状态不修改。")
public ApiResponse<TaskHeartbeatVo> interrupted(
@Parameter(description = "任务 ID", required = true, example = "200")
@PathVariable Long taskId,
@RequestBody(required = false) TaskHeartbeatRequest request) {
String reason = request == null ? null : request.getPhase();
return ApiResponse.success(taskHeartbeatService.markInterrupted(taskId, reason));
}
} }
@@ -117,6 +117,56 @@ public class TaskHeartbeatService {
return fileTaskMapper.selectOne(fileQuery); return fileTaskMapper.selectOne(fileQuery);
} }
/**
* 客户端崩溃恢复:客户端重启后发现自己上次进程级崩溃(journal 残留),
* 调此接口把上次还在 RUNNING 的任务立即标终态,替代最长等 30 分钟的 stale 兜底。
* file_task RUNNING → FAILEDbrand_crawl_tasks running → cancelled;其余状态不动(幂等)。
*/
public TaskHeartbeatVo markInterrupted(Long taskId, String reason) {
if (taskId == null || taskId <= 0) {
log.warn("[task-interrupted] ignored invalid taskId={}", taskId);
return TaskHeartbeatVo.notAlive(null, null, "invalid taskId");
}
String safeReason = (reason == null || reason.isBlank())
? "客户端异常中断,任务已自动失败"
: "客户端异常中断: " + reason.trim();
FileTaskEntity fileTask = selectFileTask(taskId);
if (fileTask != null) {
String status = fileTask.getStatus();
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getId, taskId)
.eq(FileTaskEntity::getStatus, STATUS_RUNNING)
.set(FileTaskEntity::getStatus, "FAILED")
.set(FileTaskEntity::getErrorMessage, safeReason)
.set(FileTaskEntity::getFinishedAt, LocalDateTime.now()));
if (updated > 0) {
log.warn("[task-interrupted] file task marked failed by client restart taskId={} moduleType={} reason={}",
taskId, fileTask.getModuleType(), safeReason);
return TaskHeartbeatVo.notAlive(fileTask.getModuleType(), "FAILED", "marked failed");
}
log.info("[task-interrupted] file task not in RUNNING, skipped taskId={} status={}", taskId, status);
return TaskHeartbeatVo.notAlive(fileTask.getModuleType(), status, "task is not running");
}
BrandCrawlTaskEntity brandTask = selectBrandTask(taskId);
if (brandTask != null) {
String status = brandTask.getStatus();
if ("running".equalsIgnoreCase(status) || "pending".equalsIgnoreCase(status)) {
brandTask.setStatus("cancelled");
brandTask.setErrorMessage(safeReason);
int updated = brandCrawlTaskMapper.updateById(brandTask);
if (updated > 0) {
log.warn("[task-interrupted] brand task marked cancelled by client restart taskId={} reason={}",
taskId, safeReason);
return TaskHeartbeatVo.notAlive(MODULE_BRAND, "cancelled", "marked cancelled");
}
}
log.info("[task-interrupted] brand task not in running/pending, skipped taskId={} status={}", taskId, status);
return TaskHeartbeatVo.notAlive(MODULE_BRAND, status, "task is not running");
}
log.warn("[task-interrupted] task not found taskId={}", taskId);
return TaskHeartbeatVo.notAlive(null, null, "task not found");
}
private BrandCrawlTaskEntity selectBrandTask(Long taskId) { private BrandCrawlTaskEntity selectBrandTask(Long taskId) {
LambdaQueryWrapper<BrandCrawlTaskEntity> brandQuery = new LambdaQueryWrapper<BrandCrawlTaskEntity>() LambdaQueryWrapper<BrandCrawlTaskEntity> brandQuery = new LambdaQueryWrapper<BrandCrawlTaskEntity>()
.eq(BrandCrawlTaskEntity::getId, taskId) .eq(BrandCrawlTaskEntity::getId, taskId)
@@ -163,7 +163,7 @@ import {
type ApiSecretModuleKey, type ApiSecretModuleKey,
type ApiSecretRetention, type ApiSecretRetention,
} from '@/shared/utils/api-secret-store' } from '@/shared/utils/api-secret-store'
import { getPywebviewApi, type ProxyMode } from '@/shared/bridges/pywebview' import { getPywebviewApi, type ProxyMode, type DesktopConfigUpdate } from '@/shared/bridges/pywebview'
type SecretState = { type SecretState = {
value: string value: string
@@ -275,6 +275,39 @@ function clearSecret(moduleKey: ApiSecretModuleKey) {
ElMessage.success('已清空密钥') ElMessage.success('已清空密钥')
} }
function proxyUserId() {
if (typeof window === 'undefined') return '0'
return window.localStorage.getItem('uid') || '0'
}
/** 代理地址按登录用户隔离:proxy_users[uid],各自计费各自复用;
* 未登录(uid=0)回退全局 proxy_url(兼容旧版本已保存的配置)。 */
function readUserProxy(config: Record<string, unknown> | null | undefined) {
const uid = proxyUserId()
if (uid !== '0') {
const users = (config?.proxy_users ?? {}) as Record<string, unknown>
const own = (users[uid] ?? {}) as Record<string, unknown>
return {
url: typeof own.proxy_url === 'string' ? own.proxy_url : '',
mode: Number(own.proxy_mode) === 2 ? 2 : 1,
}
}
return {
url: typeof config?.proxy_url === 'string' ? config.proxy_url : '',
mode: Number(config?.proxy_mode) === 2 ? 2 : 1,
}
}
function userProxyPatch(nextProxyUrl: string, nextProxyMode: ProxyMode) {
const uid = proxyUserId()
if (uid === '0') {
return { proxy_url: nextProxyUrl, proxy_mode: nextProxyMode }
}
const users: Record<string, unknown> = {}
users[uid] = { proxy_url: nextProxyUrl, proxy_mode: nextProxyMode }
return { proxy_users: users }
}
async function loadProxyConfig() { async function loadProxyConfig() {
const requestId = ++proxyLoadRequestId const requestId = ++proxyLoadRequestId
proxyLoading.value = true proxyLoading.value = true
@@ -294,10 +327,9 @@ async function loadProxyConfig() {
try { try {
const config = await api.read_config() const config = await api.read_config()
if (requestId !== proxyLoadRequestId || !dialogVisible.value) return if (requestId !== proxyLoadRequestId || !dialogVisible.value) return
const nextUrl = typeof config?.proxy_url === 'string' ? config.proxy_url : '' const own = readUserProxy(config as Record<string, unknown>)
const nextMode: ProxyMode = Number(config?.proxy_mode) === 2 ? 2 : 1 proxyUrl.value = own.url
proxyUrl.value = nextUrl proxyMode.value = (own.mode === 2 ? 2 : 1) as ProxyMode
proxyMode.value = nextMode
proxyReady.value = true proxyReady.value = true
} catch (error) { } catch (error) {
if (requestId !== proxyLoadRequestId || !dialogVisible.value) return if (requestId !== proxyLoadRequestId || !dialogVisible.value) return
@@ -331,10 +363,8 @@ async function saveAll() {
if (shouldSaveProxy) { if (shouldSaveProxy) {
const api = getPywebviewApi() const api = getPywebviewApi()
if (!api?.save_config) throw new Error('当前客户端未提供代理配置保存能力') if (!api?.save_config) throw new Error('当前客户端未提供代理配置保存能力')
await api.save_config({ // 按登录用户保存:proxy_users[uid](未登录回退全局 proxy_url 字段)
proxy_url: nextProxyUrl, await api.save_config(userProxyPatch(nextProxyUrl, nextProxyMode) as DesktopConfigUpdate)
proxy_mode: nextProxyMode,
})
proxyUrl.value = nextProxyUrl proxyUrl.value = nextProxyUrl
proxyDirty.value = false proxyDirty.value = false
} }
@@ -333,6 +333,17 @@ async function submitLogin() {
} catch { } catch {
/* 忽略 */ /* 忽略 */
} }
// 同步登录用户到桌面客户端配置:Python 端按 uid 选用该用户自己的
// 代理密钥/代理池(各自计费、各自复用省钱)。网页形态无 pywebview
// 桥,静默跳过。
try {
const bridge = (window as unknown as { pywebview?: { api?: { save_config?: (data: Record<string, unknown>) => Promise<unknown> } } }).pywebview
if (bridge?.api && typeof bridge.api.save_config === 'function') {
void bridge.api.save_config({ current_uid: String(data.userId) })
}
} catch {
/* 忽略 */
}
} }
if (data.username) { if (data.username) {
try { try {
@@ -365,6 +376,15 @@ onMounted(() => {
} catch { } catch {
/* 忽略 */ /* 忽略 */
} }
// 同步清掉客户端的登录用户标记:Python 端回退到全局/默认代理池
try {
const bridge = (window as unknown as { pywebview?: { api?: { save_config?: (data: Record<string, unknown>) => Promise<unknown> } } }).pywebview
if (bridge?.api && typeof bridge.api.save_config === 'function') {
void bridge.api.save_config({ current_uid: '' })
}
} catch {
/* 忽略 */
}
// 显式登出/切号后不再自动登录(保留记住的账号,仅关闭自动登录) // 显式登出/切号后不再自动登录(保留记住的账号,仅关闭自动登录)
autoLogin.value = false autoLogin.value = false
lsSet(AUTO_LOGIN_KEY, '0') lsSet(AUTO_LOGIN_KEY, '0')
@@ -56,6 +56,8 @@ export interface DesktopConfig {
export interface DesktopConfigUpdate { export interface DesktopConfigUpdate {
proxy_url?: string; proxy_url?: string;
proxy_mode?: ProxyMode; proxy_mode?: ProxyMode;
current_uid?: string;
proxy_users?: Record<string, { proxy_url?: string; proxy_mode?: number }>;
aliprice_usename?: string; aliprice_usename?: string;
aliprice_pwd?: string; aliprice_pwd?: string;
} }
@@ -8,6 +8,7 @@
import { computed, ref } from 'vue' import { computed, ref } from 'vue'
import { getPywebviewApi } from '@/shared/bridges/pywebview' import { getPywebviewApi } from '@/shared/bridges/pywebview'
import { compareVersions, normVersion } from '@/shared/utils/version-compare'
import { import {
normalizeUpdateProgress, normalizeUpdateProgress,
type UpdateProgress, type UpdateProgress,
@@ -32,10 +33,6 @@ function bindProgressListener() {
}) })
} }
function normVersion(value: unknown): string {
return String(value ?? '').trim().replace(/^[vV]/, '')
}
async function fetchJson(url: string): Promise<Record<string, unknown>> { async function fetchJson(url: string): Promise<Record<string, unknown>> {
const resp = await window.fetch(url, { credentials: 'same-origin' }) const resp = await window.fetch(url, { credentials: 'same-origin' })
if (!resp.ok) throw new Error(`HTTP ${resp.status}`) if (!resp.ok) throw new Error(`HTTP ${resp.status}`)
@@ -116,10 +113,15 @@ export function useVersionUpdate() {
: `线上最新版本 v${latest.version},暂无下载地址` : `线上最新版本 v${latest.version},暂无下载地址`
return return
} }
hasUpdate.value = local !== latest.version // 只有线上版本严格高于本机版本才算"有更新":本机领先(灰度/回滚/漏发版)
// 时不能反过来提示更新,否则会把用户往低版本上引导。
const diff = compareVersions(latest.version, local)
hasUpdate.value = diff > 0
canDownload.value = hasUpdate.value && Boolean(latest.fileUrl) canDownload.value = hasUpdate.value && Boolean(latest.fileUrl)
hint.value = hasUpdate.value hint.value = hasUpdate.value
? `发现新版本 v${latest.version}${latest.fileUrl ? '' : ',暂无下载地址'}` ? `发现新版本 v${latest.version}${latest.fileUrl ? '' : ',暂无下载地址'}`
: diff < 0
? `当前版本 v${local} 高于线上发布版本 v${latest.version},无需更新`
: `已是最新版本 v${local}` : `已是最新版本 v${local}`
} catch { } catch {
hasUpdate.value = false hasUpdate.value = false
@@ -0,0 +1,26 @@
/**
* 客户端版本号归一化与比较(桌面端更新检测用)。
*
* 只比数字段、缺失段按 0 补齐(3.0.7 == 3.0.7.0);非纯数字段(如
* 3.0.71-beta 的 beta)按 0 处理,避免 NaN 让比较结果乱序。
*/
/** 归一化版本号:去空白、去前缀 v/V,便于 'v3.0.71' 与 '3.0.71' 相等比较。 */
export function normVersion(value: unknown): string {
return String(value ?? '').trim().replace(/^[vV]/, '')
}
/** 返回 >0 表示 a 比 b 新,<0 表示 a 比 b 旧,0 表示相同。 */
export function compareVersions(a: string, b: string): number {
const pa = normVersion(a).split('.')
const pb = normVersion(b).split('.')
const len = Math.max(pa.length, pb.length)
for (let i = 0; i < len; i++) {
const na = parseInt(pa[i] ?? '', 10)
const nb = parseInt(pb[i] ?? '', 10)
const va = Number.isFinite(na) ? na : 0
const vb = Number.isFinite(nb) ? nb : 0
if (va !== vb) return va - vb
}
return 0
}
@@ -0,0 +1,39 @@
import { test } from 'node:test'
import assert from 'node:assert/strict'
import { compareVersions, normVersion } from '../src/shared/utils/version-compare.ts'
test('版本比较:线上高于本地时返回正数(应提示更新)', () => {
assert.ok(compareVersions('3.0.71', '3.0.70') > 0)
assert.ok(compareVersions('3.1.0', '3.0.71') > 0)
})
test('版本比较:线上低于本地时返回负数(不得提示更新)', () => {
// 现网真实故障场景:本地已升到 3.0.71,线上发布记录仍是 3.0.70
// 旧逻辑 local !== latest 会误报"发现新版本 v3.0.70"并诱导降级。
assert.ok(compareVersions('3.0.70', '3.0.71') < 0)
assert.ok(compareVersions('2.9.9', '3.0.0') < 0)
})
test('版本比较:完全相同(含 v 前缀、段数补齐)返回 0', () => {
assert.equal(compareVersions('3.0.71', '3.0.71'), 0)
assert.equal(compareVersions('v3.0.71', '3.0.71'), 0)
assert.equal(compareVersions('3.0.7', '3.0.7.0'), 0)
})
test('版本比较:按数字段而非字符串比较(3.0.9 < 3.0.10', () => {
assert.ok(compareVersions('3.0.9', '3.0.10') < 0)
assert.ok(compareVersions('3.0.10', '3.0.9') > 0)
})
test('版本比较:非数字段按 0 处理,不得产生 NaN 乱序', () => {
assert.equal(compareVersions('3.0.71-beta', '3.0.71'), 0)
assert.ok(compareVersions('3.0.72', '3.0.71-beta') > 0)
})
test('版本归一化:去空白与 v/V 前缀', () => {
assert.equal(normVersion(' v3.0.71 '), '3.0.71')
assert.equal(normVersion('V3.0.71'), '3.0.71')
assert.equal(normVersion(null), '')
assert.equal(normVersion(undefined), '')
})