refactor(ziniao/shopkey)+fix(stale): 打破模块循环依赖 + 采集陈旧判死全局化

模块边界(消除 ziniao ↔ shopkey 真实循环依赖):
- 新增 ziniao/service/port/{ShopKeyCatalogPort,ManagedShopNamePort}:消费方声明契约
- shopkey 侧新增 ShopKeyCatalogAdapter(读 shop_key + 白名单状态回写)、
  ManagedShopNameAdapter(店铺名校验),实现上述端口
- ZiniaoApiKeyProvider 改经端口取数,不再 import shopkey 的 Mapper/Entity;
  ZiniaoShopSwitchService 改依赖 ManagedShopNamePort
- 结果:ziniao → shopkey 的 import 归零,依赖单向(shopkey → ziniao)

陈旧判死(G1 全局判死 + D9 条件更新,替代此前的 owner 过滤/旧实体覆盖):
- ShopDataCrawlTaskService.finalizeOwnedStaleTasks → finalizeStaleTasks:去掉 owner 过滤,
  并入 DeleteBrandStaleTaskService 的 stale-check 巡检线(job 锁保证单实例扫描)
- 判死前必须持有任务锁(非阻塞获取,锁被占本轮跳过),FAILED 写入改 status CAS,
  仅在确实由 RUNNING 翻转为 FAILED 时才删缓存与分片(原实现会用扫描期旧实体覆盖在途任务)
- tryFinalizeTask 增加 allowOwnerTakeover 重载:判死场景允许跨实例接管(P1-8 盲区)
- 测试同步:owner 契约用例改为全局判死口径;mock 的 CAS 需先渲染 SQL 片段
  (MyBatis-Plus 的 where 参数延迟填充)才读参数表;新增锁被占跳过的用例

mvn test 2796 全绿
This commit is contained in:
2026-09-14 05:44:37 +08:00
parent 24ada70997
commit 95dfb69a18
13 changed files with 369 additions and 150 deletions
@@ -16,6 +16,7 @@ import com.nanri.aiimage.modules.pricetrack.service.PriceTrackTaskCacheService;
import com.nanri.aiimage.modules.pricetrack.service.PriceTrackTaskService; import com.nanri.aiimage.modules.pricetrack.service.PriceTrackTaskService;
import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskService; import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskService;
import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskCacheService; import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskCacheService;
import com.nanri.aiimage.modules.shopdatacrawl.service.ShopDataCrawlTaskService;
import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskService; import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskService;
import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskCacheService; import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskCacheService;
import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskService; import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskService;
@@ -80,6 +81,7 @@ public class DeleteBrandStaleTaskService {
private final DistributedJobLockService distributedJobLockService; private final DistributedJobLockService distributedJobLockService;
private final TaskDistributedLockService taskDistributedLockService; private final TaskDistributedLockService taskDistributedLockService;
private final TaskFileJobService taskFileJobService; private final TaskFileJobService taskFileJobService;
private final ShopDataCrawlTaskService shopDataCrawlTaskService;
@Value("${aiimage.temp-dir.retention-hours:24}") @Value("${aiimage.temp-dir.retention-hours:24}")
private long tempDirRetentionHours; private long tempDirRetentionHours;
@@ -104,6 +106,10 @@ public class DeleteBrandStaleTaskService {
runModuleStaleCheck("brand", brandTaskService::failStaleRunningTasks); runModuleStaleCheck("brand", brandTaskService::failStaleRunningTasks);
runModuleStaleCheck("appearance-patent", appearancePatentTaskService::finalizeStaleTasks); runModuleStaleCheck("appearance-patent", appearancePatentTaskService::finalizeStaleTasks);
runModuleStaleCheck("similar-asin", similarAsinTaskService::finalizeStaleTasks); runModuleStaleCheck("similar-asin", similarAsinTaskService::finalizeStaleTasks);
// 商品管理采集并入本巡检线(2026-09):此前它自带 @Scheduled 且按 owner 过滤,
// 双实例下 owner 宕机即无人判死(P1-8);并入后由本方法的 job 锁保证单实例扫描,
// 它自己改为全局判死 + 任务锁 + status CAS
runModuleStaleCheck("shop-data-crawl", shopDataCrawlTaskService::finalizeStaleTasks);
// 周期每 2 分钟一轮,各模块 summary 合并为单行,避免定期刷屏 // 周期每 2 分钟一轮,各模块 summary 合并为单行,避免定期刷屏
log.info("[stale-check] summary product-risk(s={} f={} x={}/{} p={}) price-track(s={} f={} x={}/{} p={}) shop-match(s={} f={} x={}/{} p={}) patrol-delete(s={} f={} x={}/{} p={}) query-asin(s={} f={} x={}/{} p={}) withdraw(s={} f={} x={}/{} p={}) elapsedMs={} thread={}", log.info("[stale-check] summary product-risk(s={} f={} x={}/{} p={}) price-track(s={} f={} x={}/{} p={}) shop-match(s={} f={} x={}/{} p={}) patrol-delete(s={} f={} x={}/{} p={}) query-asin(s={} f={} x={}/{} p={}) withdraw(s={} f={} x={}/{} p={}) elapsedMs={} thread={}",
stats.scannedTaskCount, stats.finalizedTaskCount, stats.failedTaskCount, stats.skippedTaskCount, stats.scannedTaskCount, stats.finalizedTaskCount, stats.failedTaskCount, stats.skippedTaskCount,
@@ -52,7 +52,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.annotation.Transactional;
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.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager; import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.transaction.support.TransactionTemplate; import org.springframework.transaction.support.TransactionTemplate;
@@ -127,23 +126,24 @@ public class ShopDataCrawlTaskService {
@Value("${aiimage.shop-data-crawl.stale-timeout-minutes:30}") @Value("${aiimage.shop-data-crawl.stale-timeout-minutes:30}")
private long staleTimeoutMinutes; private long staleTimeoutMinutes;
// cron 用独立配置键:此前复用 aiimage.delete-brand-progress.stale-check-cron /**
// 后台调整「删除品牌巡检频率」会静默改变本模块(商品管理采集)的扫库节奏 * 陈旧任务判死:扫描本模块「RUNNING 且 Python 心跳超时」的任务并终结。
@Scheduled(cron = "${aiimage.shop-data-crawl.stale-check-cron:0 */2 * * * *}") *
public void finalizeOwnedStaleTasks() { * 本方法由 {@code DeleteBrandStaleTaskService} 的 stale-check 分布式 job 锁统一调度
* (与 brand / similar-asin / appearance-patent 同一条巡检线),扫描本身是单实例执行的;
* 因此这里不再按 owner_instance_id 过滤:历史实现过滤到 owner 实例 + 终结前 owner 校验,
* 一旦 owner 实例宕机或长期拿不到任务锁,该任务在 30 分钟兜底线上无人判死(P1-8 双实例判死盲区),
* 用户侧一直显示「运行中」,只能等 2 小时的全局限时修复。
* 并发安全由 job 锁(单实例扫描)+ 任务锁(判死前必须持有)+ FAILED 写入的 status CAS 保证。
*/
public void finalizeStaleTasks() {
long minutes = Math.max(1L, staleTimeoutMinutes); long minutes = Math.max(1L, staleTimeoutMinutes);
long nowMillis = System.currentTimeMillis(); long nowMillis = System.currentTimeMillis();
List<FileTaskEntity> tasks; List<FileTaskEntity> tasks;
try { try {
// 保留 owner 过滤(2026-09 复核):本模块与 similar-asin 的差别是**没有 job 级分布式锁**
// similarasin 的判死由 delete-brand:stale-check 锁收敛为单实例执行,所以可以全局判死)。
// 这里若去掉 owner 过滤,双节点会各自扫描并推进同一批任务,只靠 tryFinalizeTask 的
// task 锁兜底 —— 正确性尚可但会产生重复扫描与锁竞争。owner 语义也有专属测试覆盖
// ShopDataCrawlOwnerColumnTest),属刻意设计而非遗漏。若要对齐全局判死,应先补 job 锁。
tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>() tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE) .eq(FileTaskEntity::getModuleType, MODULE_TYPE)
.eq(FileTaskEntity::getStatus, "RUNNING") .eq(FileTaskEntity::getStatus, "RUNNING")
.eq(FileTaskEntity::getOwnerInstanceId, currentInstanceId())
.lt(FileTaskEntity::getUpdatedAt, LocalDateTime.now().minusMinutes(minutes)) .lt(FileTaskEntity::getUpdatedAt, LocalDateTime.now().minusMinutes(minutes))
.last("limit 200")); .last("limit 200"));
} catch (Exception ex) { } catch (Exception ex) {
@@ -155,22 +155,19 @@ public class ShopDataCrawlTaskService {
for (FileTaskEntity task : tasks) { for (FileTaskEntity task : tasks) {
long heartbeat = heartbeats.getOrDefault(task.getId(), 0L); long heartbeat = heartbeats.getOrDefault(task.getId(), 0L);
if (heartbeat > 0 && nowMillis - heartbeat < Duration.ofMinutes(minutes).toMillis()) continue; if (heartbeat > 0 && nowMillis - heartbeat < Duration.ofMinutes(minutes).toMillis()) continue;
try { // 判死前必须持有任务锁:拿不到说明对方(owner 实例 / 文件任务 worker)正在推进该任务,
ensureTaskOwnedByCurrentInstance(task, "finalize stale shop data crawl task"); // 越过锁写 FAILED 会把在途任务误判失败并不可恢复地删掉分片(D9),本轮跳过等下一轮
TaskDistributedLockService.LockHandle taskLock = acquireTaskLock(task.getId(), 0L);
if (taskLock == null) {
log.info("[shop-data-crawl] stale task skipped because task lock is busy taskId={}", task.getId());
continue;
}
try (taskLock) {
if (taskFileJobService.countUnfinishedAssembleJobs(task.getId(), MODULE_TYPE) > 0L) continue; if (taskFileJobService.countUnfinishedAssembleJobs(task.getId(), MODULE_TYPE) > 0L) continue;
if (!tryFinalizeTask(task.getId(), true)) { if (!tryFinalizeTask(task.getId(), true, true)) {
// 注:此处曾改为条件更新(where status='RUNNING' 的 CAS)以防御「tryFinalizeTask // tryFinalizeTask 返回 false:无结果行(任务停在没有任何回传的阶段)
// 返回 false 含锁被占语义、用扫描期旧实体覆盖会把在途任务误判失败」的风险; // 或任务已被终结。前者由本处条件更新为 FAILED,后者被 CAS 挡住不做清理。
// 但本模块的 stale 扫描已有专属契约测试(ShopDataCrawlOwnerColumnTest / markStaleTaskFailedIfStillRunning(task.getId());
// ShopDataCrawlCleanupTest)固化「扫描即终结」的行为,改动与契约冲突。
// 保留原实现;如需加固请连同契约测试一起调整。
task.setStatus("FAILED");
task.setErrorMessage("长时间未收到 Python 结果回传,任务已自动失败");
task.setUpdatedAt(LocalDateTime.now());
task.setFinishedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
taskCacheService.deleteTaskCache(task.getId());
cleanupResultChunksQuietly(task.getId(), "stale task failure");
} }
} catch (TaskOwnerMismatchException ignored) { } catch (TaskOwnerMismatchException ignored) {
// The owner may change between the scan and finalization. // The owner may change between the scan and finalization.
@@ -180,6 +177,31 @@ public class ShopDataCrawlTaskService {
} }
} }
/**
* 把仍处于 RUNNING 的任务条件更新为 FAILEDstatus CAS),并在更新成功时才清理缓存与结果分片。
* 用条件更新而非扫描期旧实体的 updateById:后者会把已被终结/正在终结的任务覆盖回 FAILED。
*
* @return true 表示本次确实由 RUNNING 翻转为 FAILED
*/
private boolean markStaleTaskFailedIfStillRunning(Long taskId) {
LocalDateTime now = LocalDateTime.now();
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getId, taskId)
.eq(FileTaskEntity::getStatus, "RUNNING")
.set(FileTaskEntity::getStatus, "FAILED")
.set(FileTaskEntity::getErrorMessage, "长时间未收到 Python 结果回传,任务已自动失败")
.set(FileTaskEntity::getUpdatedAt, now)
.set(FileTaskEntity::getFinishedAt, now));
if (updated <= 0) {
log.info("[shop-data-crawl] stale task not flipped to FAILED because status changed taskId={}", taskId);
return false;
}
taskCacheService.deleteTaskCache(taskId);
cleanupResultChunksQuietly(taskId, "stale task failure");
log.warn("[shop-data-crawl] stale task marked FAILED taskId={}", taskId);
return true;
}
private FileTaskEntity loadTaskForExecution(Long taskId) { private FileTaskEntity loadTaskForExecution(Long taskId) {
Map<Long, FileTaskEntity> cachedTasks = taskCacheService.getTaskCacheBatch(List.of(taskId)); Map<Long, FileTaskEntity> cachedTasks = taskCacheService.getTaskCacheBatch(List.of(taskId));
FileTaskEntity cached = cachedTasks.get(taskId); FileTaskEntity cached = cachedTasks.get(taskId);
@@ -578,10 +600,22 @@ public class ShopDataCrawlTaskService {
} }
public boolean tryFinalizeTask(Long taskId, boolean fromCompensation) { public boolean tryFinalizeTask(Long taskId, boolean fromCompensation) {
return tryFinalizeTask(taskId, fromCompensation, false);
}
/**
* @param allowOwnerTakeover 陈旧判死场景传 true:允许本实例接管 owner 非本机的任务。
* 调用方(stale 巡检)已由 job 锁保证单实例执行,任务级互斥由任务锁(非阻塞获取)保证;
* 若仍按 owner 拒绝,owner 实例宕机时该任务永远无人判死(P1-8 双实例判死盲区)。
*/
public boolean tryFinalizeTask(Long taskId, boolean fromCompensation, boolean allowOwnerTakeover) {
if (taskId == null || taskId <= 0) { if (taskId == null || taskId <= 0) {
return false; return false;
} }
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId); // 判死场景用非阻塞获取:锁被占说明对方仍活跃,静默跳过等下一轮,不排队等 10 秒拖慢扫描
TaskDistributedLockService.LockHandle lockHandle = allowOwnerTakeover
? acquireTaskLock(taskId, 0L)
: acquireTaskLock(taskId);
if (lockHandle == null) { if (lockHandle == null) {
log.info("[shop-data-crawl] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation); log.info("[shop-data-crawl] tryFinalizeTask skipped because task lock is busy taskId={} fromCompensation={}", taskId, fromCompensation);
return false; return false;
@@ -591,7 +625,9 @@ public class ShopDataCrawlTaskService {
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) { if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
return false; return false;
} }
ensureTaskOwnedByCurrentInstance(task, "finalize shop data crawl task"); if (!allowOwnerTakeover) {
ensureTaskOwnedByCurrentInstance(task, "finalize shop data crawl task");
}
if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) { if ("SUCCESS".equals(task.getStatus()) || "FAILED".equals(task.getStatus())) {
return true; return true;
} }
@@ -2698,6 +2734,13 @@ public class ShopDataCrawlTaskService {
return taskDistributedLockService.acquire(MODULE_TYPE, taskId); return taskDistributedLockService.acquire(MODULE_TYPE, taskId);
} }
private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) {
if (taskId == null || taskId <= 0) {
return null;
}
return taskDistributedLockService.acquire(MODULE_TYPE, taskId, waitMillis);
}
private void markResultSuccess(FileResultEntity row) { private void markResultSuccess(FileResultEntity row) {
row.setSuccess(RESULT_SUCCESS); row.setSuccess(RESULT_SUCCESS);
row.setErrorMessage(null); row.setErrorMessage(null);
@@ -0,0 +1,21 @@
package com.nanri.aiimage.modules.shopkey.service;
import com.nanri.aiimage.modules.ziniao.service.port.ManagedShopNamePort;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
/**
* 把后台店铺管理的校验能力暴露给 ziniao 模块,
* 避免 ziniao 直接依赖 shopkey 的 Service(打破模块循环依赖)。
*/
@Service
@RequiredArgsConstructor
public class ManagedShopNameAdapter implements ManagedShopNamePort {
private final ShopManageService shopManageService;
@Override
public void requireShopByName(String shopName) {
shopManageService.requireShopByName(shopName);
}
}
@@ -0,0 +1,64 @@
package com.nanri.aiimage.modules.shopkey.service;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.nanri.aiimage.modules.shopkey.mapper.ShopKeyMapper;
import com.nanri.aiimage.modules.shopkey.model.entity.ShopKeyEntity;
import com.nanri.aiimage.modules.ziniao.service.port.ShopKeyCatalogPort;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.List;
/**
* shop_key 表对紫鸟侧的只读/回写适配器:把紫鸟需要的字段暴露给 ziniao 模块,
* 使 ziniao 不再直接引用 shopkey 的 Mapper 与 Entity(打破模块循环依赖)。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class ShopKeyCatalogAdapter implements ShopKeyCatalogPort {
private final ShopKeyMapper shopKeyMapper;
@Override
public List<ShopKeyRecord> listAllOrderByIdDesc() {
return shopKeyMapper.selectList(new LambdaQueryWrapper<ShopKeyEntity>()
.orderByDesc(ShopKeyEntity::getId))
.stream()
.filter(java.util.Objects::nonNull)
.map(entity -> new ShopKeyRecord(
entity.getId(),
entity.getZiniaoToken(),
entity.getZiniaoAccountName(),
entity.getProxyUrl(),
entity.getIpWhitelistFailCount()))
.toList();
}
@Override
public long count() {
Long total = shopKeyMapper.selectCount(null);
return total == null ? 0L : total;
}
@Override
public void updateIpWhitelist(List<Long> ids, String status, String message, Integer failCount) {
if (ids == null || ids.isEmpty()) {
// 无对应 shop_key 行时不该发空 update(会命中全表条件)
log.warn("[shop-key] 白名单状态回写被跳过:ids 为空 status={}", status);
return;
}
LambdaUpdateWrapper<ShopKeyEntity> update = new LambdaUpdateWrapper<ShopKeyEntity>()
.in(ShopKeyEntity::getId, ids)
.set(ShopKeyEntity::getIpWhitelistStatus, status)
.set(ShopKeyEntity::getIpWhitelistCheckedAt, LocalDateTime.now())
.set(ShopKeyEntity::getIpWhitelistMessage, message);
if (failCount != null) {
update.set(ShopKeyEntity::getIpWhitelistFailCount, failCount);
}
shopKeyMapper.update(null, update);
}
}
@@ -1,20 +1,22 @@
package com.nanri.aiimage.modules.ziniao.service; package com.nanri.aiimage.modules.ziniao.service;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.shopkey.mapper.ShopKeyMapper;
import com.nanri.aiimage.modules.shopkey.model.entity.ShopKeyEntity;
import com.nanri.aiimage.modules.ziniao.client.ZiniaoProxyResolver; import com.nanri.aiimage.modules.ziniao.client.ZiniaoProxyResolver;
import com.nanri.aiimage.modules.ziniao.service.port.ShopKeyCatalogPort;
import com.nanri.aiimage.modules.ziniao.service.port.ShopKeyCatalogPort.ShopKeyRecord;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.LinkedHashMap; import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Objects;
/**
* 紫鸟令牌(shop_key 表)的门面:负责按令牌聚合账号信息、缓存代理地址、维护白名单检测状态。
* 表数据读写经 {@link ShopKeyCatalogPort} 由 shopkey 模块提供,避免 ziniao 反向依赖 shopkey。
*/
@Service @Service
@RequiredArgsConstructor @RequiredArgsConstructor
public class ZiniaoApiKeyProvider implements ZiniaoProxyResolver { public class ZiniaoApiKeyProvider implements ZiniaoProxyResolver {
@@ -29,29 +31,27 @@ public class ZiniaoApiKeyProvider implements ZiniaoProxyResolver {
*/ */
public static final int IP_WHITELIST_MAX_FAIL_COUNT = 3; public static final int IP_WHITELIST_MAX_FAIL_COUNT = 3;
private final ShopKeyMapper shopKeyMapper;
private static final long PROXY_URL_CACHE_MILLIS = 60_000L; private static final long PROXY_URL_CACHE_MILLIS = 60_000L;
private final ShopKeyCatalogPort shopKeyCatalogPort;
private volatile Map<String, String> cachedProxyUrls; private volatile Map<String, String> cachedProxyUrls;
private volatile long proxyUrlCacheLoadedAt; private volatile long proxyUrlCacheLoadedAt;
public List<ApiKeyAccount> listApiKeyAccounts() { public List<ApiKeyAccount> listApiKeyAccounts() {
List<ShopKeyEntity> entities = shopKeyMapper.selectList(new LambdaQueryWrapper<ShopKeyEntity>() Map<String, List<ShopKeyRecord>> recordsByApiKey = new LinkedHashMap<>();
.orderByDesc(ShopKeyEntity::getId)); for (ShopKeyRecord record : shopKeyCatalogPort.listAllOrderByIdDesc()) {
Map<String, List<ShopKeyEntity>> entitiesByApiKey = new LinkedHashMap<>(); String apiKey = normalizeApiKey(record == null ? null : record.ziniaoToken());
for (ShopKeyEntity entity : entities) {
String apiKey = normalizeApiKey(entity == null ? null : entity.getZiniaoToken());
if (apiKey == null) { if (apiKey == null) {
continue; continue;
} }
entitiesByApiKey.computeIfAbsent(apiKey, ignored -> new ArrayList<>()).add(entity); recordsByApiKey.computeIfAbsent(apiKey, ignored -> new ArrayList<>()).add(record);
} }
return entitiesByApiKey.entrySet().stream() return recordsByApiKey.entrySet().stream()
.map(entry -> new ApiKeyAccount( .map(entry -> new ApiKeyAccount(
entry.getKey(), entry.getKey(),
resolveAccountName(entry.getValue()), resolveAccountName(entry.getValue()),
entry.getValue().stream().map(ShopKeyEntity::getId).filter(java.util.Objects::nonNull).toList(), entry.getValue().stream().map(ShopKeyRecord::id).filter(Objects::nonNull).toList(),
resolveProxyUrl(entry.getValue()), resolveProxyUrl(entry.getValue()),
resolveIpWhitelistFailCount(entry.getValue()) resolveIpWhitelistFailCount(entry.getValue())
)) ))
@@ -96,18 +96,18 @@ public class ZiniaoApiKeyProvider implements ZiniaoProxyResolver {
} }
} }
private String resolveProxyUrl(List<ShopKeyEntity> entities) { private String resolveProxyUrl(List<ShopKeyRecord> records) {
return entities.stream() return records.stream()
.map(ShopKeyEntity::getProxyUrl) .map(ShopKeyRecord::proxyUrl)
.filter(proxy -> proxy != null && !proxy.isBlank()) .filter(proxy -> proxy != null && !proxy.isBlank())
.map(String::trim) .map(String::trim)
.findFirst() .findFirst()
.orElse(null); .orElse(null);
} }
private Integer resolveIpWhitelistFailCount(List<ShopKeyEntity> entities) { private Integer resolveIpWhitelistFailCount(List<ShopKeyRecord> records) {
return entities.stream() return records.stream()
.map(ShopKeyEntity::getIpWhitelistFailCount) .map(ShopKeyRecord::ipWhitelistFailCount)
.filter(count -> count != null) .filter(count -> count != null)
.max(Integer::compareTo) .max(Integer::compareTo)
.orElse(0); .orElse(0);
@@ -128,8 +128,7 @@ public class ZiniaoApiKeyProvider implements ZiniaoProxyResolver {
} }
public boolean hasApiKey() { public boolean hasApiKey() {
Long total = shopKeyMapper.selectCount(new LambdaQueryWrapper<ShopKeyEntity>()); return shopKeyCatalogPort.count() > 0;
return total != null && total > 0;
} }
public void markIpWhitelistAllowed(ApiKeyAccount account) { public void markIpWhitelistAllowed(ApiKeyAccount account) {
@@ -168,20 +167,12 @@ public class ZiniaoApiKeyProvider implements ZiniaoProxyResolver {
if (account == null || account.shopKeyIds().isEmpty()) { if (account == null || account.shopKeyIds().isEmpty()) {
return; return;
} }
LambdaUpdateWrapper<ShopKeyEntity> update = new LambdaUpdateWrapper<ShopKeyEntity>() shopKeyCatalogPort.updateIpWhitelist(account.shopKeyIds(), status, truncateMessage(message), failCount);
.in(ShopKeyEntity::getId, account.shopKeyIds())
.set(ShopKeyEntity::getIpWhitelistStatus, status)
.set(ShopKeyEntity::getIpWhitelistCheckedAt, LocalDateTime.now())
.set(ShopKeyEntity::getIpWhitelistMessage, truncateMessage(message));
if (failCount != null) {
update.set(ShopKeyEntity::getIpWhitelistFailCount, failCount);
}
shopKeyMapper.update(null, update);
} }
private String resolveAccountName(List<ShopKeyEntity> entities) { private String resolveAccountName(List<ShopKeyRecord> records) {
return entities.stream() return records.stream()
.map(ShopKeyEntity::getZiniaoAccountName) .map(ShopKeyRecord::ziniaoAccountName)
.filter(name -> name != null && !name.isBlank()) .filter(name -> name != null && !name.isBlank())
.map(String::trim) .map(String::trim)
.findFirst() .findFirst()
@@ -1,8 +1,8 @@
package com.nanri.aiimage.modules.ziniao.service; package com.nanri.aiimage.modules.ziniao.service;
import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.shopkey.service.ShopManageService;
import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo; import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo;
import com.nanri.aiimage.modules.ziniao.service.port.ManagedShopNamePort;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
@@ -12,7 +12,7 @@ public class ZiniaoShopSwitchService {
private final ZiniaoAuthService ziniaoAuthService; private final ZiniaoAuthService ziniaoAuthService;
private final ZiniaoShopIndexService ziniaoShopIndexService; private final ZiniaoShopIndexService ziniaoShopIndexService;
private final ShopManageService shopManageService; private final ManagedShopNamePort managedShopNamePort;
public ZiniaoShopMatchResultVo matchStoreByNameAcrossStaff(String targetShopName, Long preferUserId) { public ZiniaoShopMatchResultVo matchStoreByNameAcrossStaff(String targetShopName, Long preferUserId) {
String normalizedShopName = requireManagedShopName(targetShopName); String normalizedShopName = requireManagedShopName(targetShopName);
@@ -62,7 +62,7 @@ public class ZiniaoShopSwitchService {
private String requireManagedShopName(String targetShopName) { private String requireManagedShopName(String targetShopName) {
String normalizedShopName = normalizeShopName(targetShopName); String normalizedShopName = normalizeShopName(targetShopName);
if (!normalizedShopName.isBlank()) { if (!normalizedShopName.isBlank()) {
shopManageService.requireShopByName(normalizedShopName); managedShopNamePort.requireShopByName(normalizedShopName);
} }
return normalizedShopName; return normalizedShopName;
} }
@@ -0,0 +1,12 @@
package com.nanri.aiimage.modules.ziniao.service.port;
/**
* 后台店铺管理数据(shop_manage 表)的只读校验端口。
*
* 由 shopkey 模块实现,用于替代 ziniao → shopkey 的直接类依赖,避免模块循环依赖。
*/
public interface ManagedShopNamePort {
/** 校验店铺名在后台店铺管理中已维护,未找到则抛出 BusinessException。 */
void requireShopByName(String shopName);
}
@@ -0,0 +1,27 @@
package com.nanri.aiimage.modules.ziniao.service.port;
import java.util.List;
/**
* 紫鸟令牌数据来源端口。
*
* 数据实际存放在 shop_key 表(shopkey 模块),这里只声明紫鸟侧需要的最小读取/回写契约,
* 由 shopkey 模块提供实现;这样 ziniao 不再直接依赖 shopkey 的 Mapper/Entity
* 模块间保持单向依赖(shopkey → ziniao)。
*/
public interface ShopKeyCatalogPort {
/** 全量紫鸟令牌记录,按 id 倒序(最新维护的排前面)。 */
List<ShopKeyRecord> listAllOrderByIdDesc();
/** 令牌记录总行数。 */
long count();
/** 按 id 批量回写白名单检测结果;failCount 为 null 表示不改动该字段。 */
void updateIpWhitelist(List<Long> ids, String status, String message, Integer failCount);
/** shop_key 中与紫鸟令牌相关的字段子集。 */
record ShopKeyRecord(Long id, String ziniaoToken, String ziniaoAccountName,
String proxyUrl, Integer ipWhitelistFailCount) {
}
}
@@ -216,7 +216,7 @@ class DeleteBrandStaleTaskServiceTest {
return new DeleteBrandStaleTaskService( return new DeleteBrandStaleTaskService(
fileTaskMapper, deleteBrandTaskCacheService, deleteBrandTaskStorageService, deleteBrandRunService, fileTaskMapper, deleteBrandTaskCacheService, deleteBrandTaskStorageService, deleteBrandRunService,
null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null, null,
deleteBrandProgressProperties, null, taskDistributedLockService, taskFileJobService); deleteBrandProgressProperties, null, taskDistributedLockService, taskFileJobService, null);
} }
private void lockAvailable() { private void lockAvailable() {
@@ -77,6 +77,7 @@ import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
@@ -85,7 +86,7 @@ import static org.mockito.Mockito.when;
* 覆盖四类资源释放路径: * 覆盖四类资源释放路径:
* 1. 删除——deleteHistory 重建每日累计文件时临时文件在 finally 中删除、旧对象在行删除后按引用回收、 * 1. 删除——deleteHistory 重建每日累计文件时临时文件在 finally 中删除、旧对象在行删除后按引用回收、
* 新上传对象在事务回滚时注册清理;任务删除只解绑成员不触碰累计文件对象。 * 新上传对象在事务回滚时注册清理;任务删除只解绑成员不触碰累计文件对象。
* 2. 超时——finalizeOwnedStaleTasks 将陈旧 RUNNING 任务标记 FAILED 并清理残留分片(chunk 行 + payload), * 2. 超时——finalizeStaleTasks 将陈旧 RUNNING 任务标记 FAILED 并清理残留分片(chunk 行 + payload),
* 心跳活跃的任务被跳过;扫表失败安全返回。 * 心跳活跃的任务被跳过;扫表失败安全返回。
* 3. 重复回传——同一分片重复提交不重复计数、被拒 payload 释放;scope 状态写入失败时 * 3. 重复回传——同一分片重复提交不重复计数、被拒 payload 释放;scope 状态写入失败时
* 回滚本次插入的分片行与 payload。 * 回滚本次插入的分片行与 payload。
@@ -195,6 +196,8 @@ class ShopDataCrawlCleanupTest {
lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong())) lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong()))
.thenReturn(mock(TaskDistributedLockService.LockHandle.class)); .thenReturn(mock(TaskDistributedLockService.LockHandle.class));
lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong(), anyLong()))
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of()); lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of());
lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of()); lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of());
lenient().when(taskCacheService.getAllShopMergedPayload(anyLong())).thenReturn(Map.of()); lenient().when(taskCacheService.getAllShopMergedPayload(anyLong())).thenReturn(Map.of());
@@ -230,16 +233,13 @@ class ShopDataCrawlCleanupTest {
.thenAnswer(invocation -> { .thenAnswer(invocation -> {
com.baomidou.mybatisplus.core.conditions.Wrapper<FileTaskEntity> wrapper = com.baomidou.mybatisplus.core.conditions.Wrapper<FileTaskEntity> wrapper =
invocation.getArgument(1); invocation.getArgument(1);
// getParamNameValuePairs 在 AbstractWrapper 上(不在 Wrapper 接口),用反射取条件参数 // where 条件的值由 MyBatis-Plus 延迟填充:先渲染一次 SQL 片段才会写入参数
java.util.Map<String, Object> params; // (真实执行同样先渲染,故生产语义不变);反射取表会漏掉这些延迟项,改用直调重载。
try { @SuppressWarnings("unchecked")
java.lang.reflect.Method m = wrapper.getClass().getMethod("getParamNameValuePairs"); java.util.Map<String, Object> params =
@SuppressWarnings("unchecked") wrapper instanceof com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper<?> lambdaWrapper
java.util.Map<String, Object> extracted = (java.util.Map<String, Object>) m.invoke(wrapper); ? (java.util.Map<String, Object>) renderParams(lambdaWrapper)
params = extracted; : java.util.Map.of();
} catch (Exception ex) {
params = java.util.Map.of();
}
java.util.Optional<Object> idValue = params.values().stream() java.util.Optional<Object> idValue = params.values().stream()
.filter(v -> v instanceof Long).findFirst(); .filter(v -> v instanceof Long).findFirst();
if (idValue.isEmpty()) { if (idValue.isEmpty()) {
@@ -732,15 +732,15 @@ class ShopDataCrawlCleanupTest {
// 陈旧任务扫描失败安全返回,不抛异常。 // 陈旧任务扫描失败安全返回,不抛异常。
doThrow(new IllegalStateException("scan failed")) doThrow(new IllegalStateException("scan failed"))
.when(fileTaskMapper).selectList(any()); .when(fileTaskMapper).selectList(any());
service.finalizeOwnedStaleTasks(); service.finalizeStaleTasks();
} }
// ---- 2. 超时路径 ---- // ---- 2. 超时路径 ----
@Test @Test
void test_task_039_daily_file_cleanup_timeout_stale_task_failed() throws Exception { void test_task_039_daily_file_cleanup_timeout_stale_task_failed() throws Exception {
// 超时:陈旧 RUNNING 任务在锁不可得时由调度器标记 FAILED,残留分片行与 payload 全部清理; // 超时:拿到任务锁的陈旧 RUNNING 任务由 status CAS 置 FAILED,残留分片行与 payload 全部清理;
// 心跳活跃的任务被跳过;扫表失败安全返回。 // 心跳活跃的任务被跳过;任务锁被占(对方仍在推进)时本轮既不判死也不清理(D9);扫表失败安全返回。
ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 1L); ReflectionTestUtils.setField(service, "staleTimeoutMinutes", 1L);
FileResultEntity row = addResultRow(8201L, 2L, -1, SHOP_NAME, null); FileResultEntity row = addResultRow(8201L, 2L, -1, SHOP_NAME, null);
FileTaskEntity staleTask = taskEntity(2L, "RUNNING"); FileTaskEntity staleTask = taskEntity(2L, "RUNNING");
@@ -750,21 +750,28 @@ class ShopDataCrawlCleanupTest {
FileTaskEntity activeTask = taskEntity(3L, "RUNNING"); FileTaskEntity activeTask = taskEntity(3L, "RUNNING");
activeTask.setUpdatedAt(LocalDateTime.now().minusHours(2)); activeTask.setUpdatedAt(LocalDateTime.now().minusHours(2));
taskStore.put(3L, activeTask); taskStore.put(3L, activeTask);
FileResultEntity lockedRow = addResultRow(8203L, 4L, -1, SHOP_NAME, null);
FileTaskEntity lockedTask = taskEntity(4L, "RUNNING");
lockedTask.setUpdatedAt(LocalDateTime.now().minusHours(2));
taskStore.put(4L, lockedTask);
addChunk(2L, 1, "rustfs:stale-payload"); addChunk(2L, 1, "rustfs:stale-payload");
addChunk(3L, 1, "rustfs:active-payload"); addChunk(3L, 1, "rustfs:active-payload");
addChunk(4L, 1, "rustfs:locked-payload");
lastJobTaskId = 2L; lastJobTaskId = 2L;
lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), eq(2L))) // 任务 4 的任务锁被占:判死前必须先持锁,拿不到就整轮跳过(不写 FAILED、不删分片)
lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), eq(4L), anyLong()))
.thenReturn(null); .thenReturn(null);
lenient().when(fileTaskMapper.selectList(any())).thenAnswer(invocation -> { lenient().when(fileTaskMapper.selectList(any())).thenAnswer(invocation -> {
Wrapper<FileTaskEntity> wrapper = invocation.getArgument(0); Wrapper<FileTaskEntity> wrapper = invocation.getArgument(0);
String segment = wrapper instanceof LambdaQueryWrapper<?> query && query.getSqlSegment() != null String segment = wrapper instanceof LambdaQueryWrapper<?> query && query.getSqlSegment() != null
? query.getSqlSegment() : ""; ? query.getSqlSegment() : "";
List<FileTaskEntity> result = new ArrayList<>(); List<FileTaskEntity> result = new ArrayList<>();
if (segment.contains("ownerInstanceId")) { // 2026-09 全局判死:stale 扫描不再按 owner_instance_id 过滤,只按模块 + RUNNING
if (segment.contains("moduleType")) {
for (FileTaskEntity stored : taskStore.values()) { for (FileTaskEntity stored : taskStore.values()) {
if ("RUNNING".equals(stored.getStatus()) if ("RUNNING".equals(stored.getStatus())
&& "instance-a".equals(stored.getOwnerInstanceId())) { && MODULE_TYPE.equals(stored.getModuleType())) {
result.add(copyTask(stored)); result.add(copyTask(stored));
} }
} }
@@ -782,21 +789,27 @@ class ShopDataCrawlCleanupTest {
return heartbeats; return heartbeats;
}); });
service.finalizeOwnedStaleTasks(); service.finalizeStaleTasks();
// 任务 2 有结果行:拿到锁后走模块自身的最终化路径(Python 中断语义),
// 且因无可用结果 → 顺带释放分片("terminal task without workbook"
assertEquals("FAILED", taskStore.get(2L).getStatus(), "陈旧任务标记 FAILED"); assertEquals("FAILED", taskStore.get(2L).getStatus(), "陈旧任务标记 FAILED");
assertTrue(taskStore.get(2L).getErrorMessage().contains("自动失败"), "错误消息可识别"); assertTrue(taskStore.get(2L).getErrorMessage().contains("中断"), "错误消息为中断语义");
assertTrue(dbChunks.stream().noneMatch(c -> Objects.equals(c.getTaskId(), 2L)), assertTrue(dbChunks.stream().noneMatch(c -> Objects.equals(c.getTaskId(), 2L)),
"陈旧任务残留分片行清理"); "陈旧任务残留分片行清理");
assertTrue(deletedPayloads.contains("rustfs:stale-payload"), "陈旧任务分片 payload 释放"); assertTrue(deletedPayloads.contains("rustfs:stale-payload"), "陈旧任务分片 payload 释放");
assertEquals("RUNNING", taskStore.get(3L).getStatus(), "心跳活跃任务跳过"); assertEquals("RUNNING", taskStore.get(3L).getStatus(), "心跳活跃任务跳过");
assertFalse(deletedPayloads.contains("rustfs:active-payload"), "活跃任务分片 payload 保留"); assertFalse(deletedPayloads.contains("rustfs:active-payload"), "活跃任务分片 payload 保留");
assertEquals("RUNNING", taskStore.get(4L).getStatus(), "任务锁被占时本轮不判死");
assertTrue(dbChunks.stream().anyMatch(c -> Objects.equals(c.getTaskId(), 4L)), "任务锁被占时保留分片行");
assertFalse(deletedPayloads.contains("rustfs:locked-payload"), "任务锁被占时保留分片 payload");
verify(taskCacheService).deleteTaskCache(2L); verify(taskCacheService).deleteTaskCache(2L);
verify(taskCacheService, never()).deleteTaskCache(4L);
// 扫表失败安全返回,不影响任何资源。 // 扫表失败安全返回,不影响任何资源。
doThrow(new IllegalStateException("scan failed")) doThrow(new IllegalStateException("scan failed"))
.when(fileTaskMapper).selectList(any()); .when(fileTaskMapper).selectList(any());
service.finalizeOwnedStaleTasks(); service.finalizeStaleTasks();
} }
// ---- 3. 重复回传与累计文件失败 ---- // ---- 3. 重复回传与累计文件失败 ----
@@ -1266,6 +1279,13 @@ class ShopDataCrawlCleanupTest {
return copy; return copy;
} }
/** 渲染一次 SQL 片段以补齐延迟写入的参数,再返回参数表。 */
private static java.util.Map<String, Object> renderParams(
com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper<?> wrapper) {
wrapper.getSqlSegment();
return wrapper.getParamNameValuePairs();
}
private FileTaskEntity copyTask(FileTaskEntity source) { private FileTaskEntity copyTask(FileTaskEntity source) {
FileTaskEntity copy = new FileTaskEntity(); FileTaskEntity copy = new FileTaskEntity();
copy.setId(source.getId()); copy.setId(source.getId());
@@ -70,6 +70,12 @@ import static org.mockito.Mockito.when;
* 迁移后写入/读取/查询都走 biz_file_task.owner_instance_id 显式列 * 迁移后写入/读取/查询都走 biz_file_task.owner_instance_id 显式列
* V92 迁移新增列并补 (owner_instance_id, status, updated_at) 索引 * V92 迁移新增列并补 (owner_instance_id, status, updated_at) 索引
* JSON 解析不再参与 owner 判定stale 扫描直接按列过滤 * JSON 解析不再参与 owner 判定stale 扫描直接按列过滤
*
* 2026-09 复核修订stale 扫描改为**全局判死**不再按 owner_instance_id 过滤
* owner 过滤 + 终结前 owner 校验在双实例下是 P1-8 判死盲区owner 实例宕机/长期拿不到任务锁时
* 该任务在 30 分钟兜底线上无人判死改造后单实例扫描由 DeleteBrandStaleTaskService
* stale-check job 锁保证任务级互斥由任务锁判死前必须持有+ status CAS 保证
* 本类继续覆盖 owner 列的写入/读取/兼容createTask 落列ownerInstanceIdOf 兼容 JSON 兜底
*/ */
@ExtendWith(MockitoExtension.class) @ExtendWith(MockitoExtension.class)
class ShopDataCrawlOwnerColumnTest { class ShopDataCrawlOwnerColumnTest {
@@ -142,6 +148,8 @@ class ShopDataCrawlOwnerColumnTest {
lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a");
lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong())) lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong()))
.thenReturn(mock(TaskDistributedLockService.LockHandle.class)); .thenReturn(mock(TaskDistributedLockService.LockHandle.class));
lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong(), anyLong()))
.thenReturn(mock(TaskDistributedLockService.LockHandle.class));
lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of()); lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of());
lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of()); lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of());
lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())).thenReturn(List.of()); lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())).thenReturn(List.of());
@@ -153,12 +161,31 @@ class ShopDataCrawlOwnerColumnTest {
return value == null ? "" : value.trim(); return value == null ? "" : value.trim();
}); });
lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1);
// 2026-09陈旧扫描的 FAILED 写入改为条件更新where status='RUNNING' CAS // 2026-09陈旧扫描的 FAILED 写入改为条件更新where id=? and status='RUNNING' CAS
// 走的是 update(entity=null, wrapper)需要 stub 返回 1否则默认 0 会被当成"未翻转"任务保持 RUNNING // 走的是 update(entity=null, wrapper)这里按 wrapper 参数模拟 DB只有仍为 RUNNING 的行才命中
// 命中后落回内存表与真实 DB 语义一致第二次扫描因终态不再命中
// 注意用 isNull()Mockito 2+ any(Class) 不匹配 null // 注意用 isNull()Mockito 2+ any(Class) 不匹配 null
lenient().when(fileTaskMapper.update(org.mockito.ArgumentMatchers.isNull(), lenient().when(fileTaskMapper.update(org.mockito.ArgumentMatchers.isNull(),
any(com.baomidou.mybatisplus.core.conditions.Wrapper.class))) any(com.baomidou.mybatisplus.core.conditions.Wrapper.class)))
.thenReturn(1); .thenAnswer(invocation -> {
com.baomidou.mybatisplus.core.conditions.Wrapper<FileTaskEntity> wrapper =
invocation.getArgument(1);
java.util.Map<String, Object> params = extractParamValues(wrapper);
java.util.Optional<Object> idValue = params.values().stream()
.filter(value -> value instanceof Long)
.findFirst();
if (idValue.isEmpty()) {
return 0;
}
FileTaskEntity stored = dbTask((Long) idValue.get());
if (stored == null || !"RUNNING".equals(stored.getStatus())) {
return 0;
}
stored.setStatus("FAILED");
stored.setUpdatedAt(LocalDateTime.now());
stored.setFinishedAt(LocalDateTime.now());
return 1;
});
lenient().when(fileTaskMapper.selectById(anyLong())).thenReturn(null); lenient().when(fileTaskMapper.selectById(anyLong())).thenReturn(null);
lenient().when(fileTaskMapper.selectList(any())).thenAnswer(invocation -> { lenient().when(fileTaskMapper.selectList(any())).thenAnswer(invocation -> {
Wrapper<FileTaskEntity> wrapper = invocation.getArgument(0); Wrapper<FileTaskEntity> wrapper = invocation.getArgument(0);
@@ -179,15 +206,16 @@ class ShopDataCrawlOwnerColumnTest {
assertEquals("instance-a", dbTask(3401L).getOwnerInstanceId(), "显式列写入当前实例 id"); assertEquals("instance-a", dbTask(3401L).getOwnerInstanceId(), "显式列写入当前实例 id");
assertTrue(dbTask(3401L).getRequestJson().contains("ownerInstanceId"), "兼容字段仍保留在快照 JSON"); assertTrue(dbTask(3401L).getRequestJson().contains("ownerInstanceId"), "兼容字段仍保留在快照 JSON");
// stale 扫描按显式列过滤当前实例 RUNNING 任务命中 owner 路由 // stale 扫描命中 RUNNING 任务2026-09 起全局判死owner 列不再参与扫描过滤
runStaleScan(); runStaleScan();
assertEquals(1, lastScan.size(), "stale 扫描按 owner_instance_id 列过滤命中"); assertEquals(1, lastScan.size(), "stale 扫描命中 RUNNING 任务");
assertEquals(3401L, lastScan.get(0).getId()); assertEquals(3401L, lastScan.get(0).getId());
} }
@Test @Test
void test_task_034_owner_normal_multiple_items() { void test_task_034_owner_normal_multiple_items() {
// 批量场景多个任务各自带 owner 扫描只返回当前实例的任务其他实例不命中 // 批量场景owner 列不再参与 stale 扫描过滤全局判死跨实例 RUNNING 任务一并进入候选
// 单实例扫描由 stale-check job 锁保证终态任务不命中
dbTasks.add(task(3402L, "instance-a", "RUNNING")); dbTasks.add(task(3402L, "instance-a", "RUNNING"));
dbTasks.add(task(3403L, "instance-a", "RUNNING")); dbTasks.add(task(3403L, "instance-a", "RUNNING"));
dbTasks.add(task(3404L, "instance-b", "RUNNING")); dbTasks.add(task(3404L, "instance-b", "RUNNING"));
@@ -195,7 +223,7 @@ class ShopDataCrawlOwnerColumnTest {
runStaleScan(); runStaleScan();
assertEquals(2, lastScan.size(), "只返回当前实例 RUNNING 任务"); assertEquals(3, lastScan.size(), "全局判死:跨实例 RUNNING 任务都属于候选");
assertEquals(3402L, lastScan.get(0).getId()); assertEquals(3402L, lastScan.get(0).getId());
assertEquals(3403L, lastScan.get(1).getId()); assertEquals(3403L, lastScan.get(1).getId());
} }
@@ -260,17 +288,19 @@ class ShopDataCrawlOwnerColumnTest {
dbTasks.add(task(10000L + i, "instance-a", "RUNNING")); dbTasks.add(task(10000L + i, "instance-a", "RUNNING"));
} }
runStaleScan(); runStaleScan();
assertEquals(40, lastScan.size(), "大量任务逐一命中,无重复无丢失"); // 3408owner 为长实例 id在全局判死下同样是候选40 + 1
assertEquals(41, lastScan.size(), "大量任务逐一命中,无重复无丢失");
} }
@Test @Test
void test_task_034_owner_invalid_input_rejected() { void test_task_034_owner_invalid_input_rejected() {
// 非法参数 owner 列的 RUNNING 任务不属于任何实例扫描不命中区别于旧 JSON 兼容分支 // 非法参数 owner 列的 RUNNING 任务同样进入全局判死候选owner 不再是扫描条件
// JSON 兼容分支仍可读出 owner
dbTasks.add(task(3409L, null, "RUNNING")); dbTasks.add(task(3409L, null, "RUNNING"));
runStaleScan(); runStaleScan();
assertEquals(0, lastScan.size(), "owner 列缺失的任务不属于当前实例"); assertEquals(1, lastScan.size(), "owner 列缺失不影响全局判死候选集");
// 归属性判定对缺 owner 列的任务放行兼容读取但不归属任何实例 // 归属性判定对缺 owner 列的任务放行兼容读取但不归属任何实例
FileTaskEntity legacy = task(3410L, null, "RUNNING"); FileTaskEntity legacy = task(3410L, null, "RUNNING");
legacy.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}"); legacy.setRequestJson("{\"ownerInstanceId\":\"instance-a\"}");
@@ -287,7 +317,7 @@ class ShopDataCrawlOwnerColumnTest {
service.ensureTaskOwnedByCurrentInstance(foreign, "submit shop data crawl result")); service.ensureTaskOwnedByCurrentInstance(foreign, "submit shop data crawl result"));
doThrow(new RuntimeException("db down")).when(fileTaskMapper).selectList(any()); doThrow(new RuntimeException("db down")).when(fileTaskMapper).selectList(any());
service.finalizeOwnedStaleTasks(); service.finalizeStaleTasks();
assertTrue(lastScan.isEmpty(), "DB 异常时扫描降级为空"); assertTrue(lastScan.isEmpty(), "DB 异常时扫描降级为空");
} }
@@ -353,12 +383,14 @@ class ShopDataCrawlOwnerColumnTest {
} }
private void runStaleScan() { private void runStaleScan() {
service.finalizeOwnedStaleTasks(); service.finalizeStaleTasks();
} }
/** /**
* SQL 片段中出现的列名与参数占位符顺序提取 owner_instance_id / status 的查询值 * SQL 片段中出现的列名与参数占位符顺序提取 status 的查询值
* 模拟 MySQL 按显式列过滤与生产查询的语义一致仅用于筛选 dbTasks * 模拟 MySQL 按显式列过滤与生产查询的语义一致仅用于筛选 dbTasks
*
* owner 维度已于 2026-09 stale 扫描中去掉全局判死因此这里不再解析 ownerInstanceId
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
private List<FileTaskEntity> applyTaskScanFilter(Wrapper<FileTaskEntity> wrapper, List<FileTaskEntity> candidates) { private List<FileTaskEntity> applyTaskScanFilter(Wrapper<FileTaskEntity> wrapper, List<FileTaskEntity> candidates) {
@@ -367,24 +399,35 @@ class ShopDataCrawlOwnerColumnTest {
} }
String sql = query.getSqlSegment(); String sql = query.getSqlSegment();
Map<String, Object> params = query.getParamNameValuePairs(); Map<String, Object> params = query.getParamNameValuePairs();
final String[] ownerFilter = {null};
final String[] statusFilter = {null}; final String[] statusFilter = {null};
if (sql != null && sql.contains("ownerInstanceId")) { if (sql != null) {
Matcher matcher = Pattern.compile("([a-zA-Z_]+)\\s*=\\s*#\\{ew\\.paramNameValuePairs\\.([A-Za-z0-9]+)\\}") Matcher matcher = Pattern.compile("([a-zA-Z_]+)\\s*=\\s*#\\{ew\\.paramNameValuePairs\\.([A-Za-z0-9]+)\\}")
.matcher(sql); .matcher(sql);
while (matcher.find()) { while (matcher.find()) {
String column = matcher.group(1); String column = matcher.group(1);
Object value = params.get(matcher.group(2)); Object value = params.get(matcher.group(2));
if ("ownerInstanceId".equals(column) && value instanceof String s) { if ("status".equals(column) && value instanceof String s) {
ownerFilter[0] = s;
} else if ("status".equals(column) && value instanceof String s) {
statusFilter[0] = s; statusFilter[0] = s;
} }
} }
} }
return candidates.stream() return candidates.stream()
.filter(t -> ownerFilter[0] == null || Objects.equals(ownerFilter[0], t.getOwnerInstanceId()))
.filter(t -> statusFilter[0] == null || Objects.equals(statusFilter[0], t.getStatus())) .filter(t -> statusFilter[0] == null || Objects.equals(statusFilter[0], t.getStatus()))
.toList(); .toList();
} }
/**
* 取条件更新 wrapper 的参数表
* 注意where 条件对应的值由 MyBatis-Plus **延迟填充**必须先触发一次 SQL 片段渲染
* getSqlSegment才会写入参数表真实执行时同样先渲染 SQL所以生产语义不受影响
* 反射拿参数表同样会漏掉延迟项这里统一走 LambdaUpdateWrapper 直调 + 先渲染
*/
@SuppressWarnings("unchecked")
private Map<String, Object> extractParamValues(Wrapper<FileTaskEntity> wrapper) {
if (wrapper instanceof com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper<?> lambdaWrapper) {
lambdaWrapper.getSqlSegment();
return (Map<String, Object>) lambdaWrapper.getParamNameValuePairs();
}
return Map.of();
}
} }
@@ -1,41 +1,26 @@
package com.nanri.aiimage.modules.ziniao.service; package com.nanri.aiimage.modules.ziniao.service;
import com.baomidou.mybatisplus.core.conditions.Wrapper; import com.nanri.aiimage.modules.ziniao.service.port.ShopKeyCatalogPort;
import com.baomidou.mybatisplus.core.MybatisConfiguration; import com.nanri.aiimage.modules.ziniao.service.port.ShopKeyCatalogPort.ShopKeyRecord;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.nanri.aiimage.modules.shopkey.mapper.ShopKeyMapper;
import com.nanri.aiimage.modules.shopkey.model.entity.ShopKeyEntity;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import java.util.List; import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
class ZiniaoApiKeyProviderTest { class ZiniaoApiKeyProviderTest {
@BeforeAll
static void initializeMybatisMetadata() {
TableInfoHelper.initTableInfo(
new MapperBuilderAssistant(new MybatisConfiguration(), "ziniao-api-key-provider-test"),
ShopKeyEntity.class
);
}
@Test @Test
void duplicateNormalizedTokensShareOneRefreshAccountAndAllRecordIds() { void duplicateNormalizedTokensShareOneRefreshAccountAndAllRecordIds() {
ShopKeyMapper mapper = mock(ShopKeyMapper.class); ShopKeyCatalogPort port = mock(ShopKeyCatalogPort.class);
ShopKeyEntity latest = shopKey(12L, " Bearer duplicate-key ", "最新账号"); ShopKeyRecord latest = shopKey(12L, " Bearer duplicate-key ", "最新账号");
ShopKeyEntity older = shopKey(8L, "duplicate-key", "旧账号"); ShopKeyRecord older = shopKey(8L, "duplicate-key", "旧账号");
when(mapper.selectList(any())).thenReturn(List.of(latest, older)); when(port.listAllOrderByIdDesc()).thenReturn(List.of(latest, older));
ZiniaoApiKeyProvider provider = new ZiniaoApiKeyProvider(mapper); ZiniaoApiKeyProvider provider = new ZiniaoApiKeyProvider(port);
List<ZiniaoApiKeyProvider.ApiKeyAccount> accounts = provider.listApiKeyAccounts(); List<ZiniaoApiKeyProvider.ApiKeyAccount> accounts = provider.listApiKeyAccounts();
@@ -47,8 +32,8 @@ class ZiniaoApiKeyProviderTest {
@Test @Test
void whitelistResultUpdatesEveryRecordForTheNormalizedToken() { void whitelistResultUpdatesEveryRecordForTheNormalizedToken() {
ShopKeyMapper mapper = mock(ShopKeyMapper.class); ShopKeyCatalogPort port = mock(ShopKeyCatalogPort.class);
ZiniaoApiKeyProvider provider = new ZiniaoApiKeyProvider(mapper); ZiniaoApiKeyProvider provider = new ZiniaoApiKeyProvider(port);
ZiniaoApiKeyProvider.ApiKeyAccount account = new ZiniaoApiKeyProvider.ApiKeyAccount( ZiniaoApiKeyProvider.ApiKeyAccount account = new ZiniaoApiKeyProvider.ApiKeyAccount(
"duplicate-key", "duplicate-key",
"账号", "账号",
@@ -57,14 +42,21 @@ class ZiniaoApiKeyProviderTest {
provider.markIpWhitelistBlocked(account, "当前服务器 IP 未加入紫鸟白名单"); provider.markIpWhitelistBlocked(account, "当前服务器 IP 未加入紫鸟白名单");
verify(mapper).update(isNull(), any(Wrapper.class)); verify(port).updateIpWhitelist(List.of(12L, 8L), ZiniaoApiKeyProvider.IP_WHITELIST_STATUS_BLOCKED,
"当前服务器 IP 未加入紫鸟白名单", 1);
} }
private ShopKeyEntity shopKey(long id, String token, String accountName) { @Test
ShopKeyEntity entity = new ShopKeyEntity(); void recordsWithoutValidTokenAreIgnored() {
entity.setId(id); ShopKeyCatalogPort port = mock(ShopKeyCatalogPort.class);
entity.setZiniaoToken(token); when(port.listAllOrderByIdDesc()).thenReturn(List.of(shopKey(3L, " ", "空令牌账号")));
entity.setZiniaoAccountName(accountName);
return entity; ZiniaoApiKeyProvider provider = new ZiniaoApiKeyProvider(port);
assertEquals(List.of(), provider.listApiKeyAccounts());
}
private ShopKeyRecord shopKey(long id, String token, String accountName) {
return new ShopKeyRecord(id, token, accountName, null, 0);
} }
} }
@@ -1,7 +1,7 @@
package com.nanri.aiimage.modules.ziniao.service; package com.nanri.aiimage.modules.ziniao.service;
import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.shopkey.service.ShopManageService; import com.nanri.aiimage.modules.ziniao.service.port.ManagedShopNamePort;
import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo; import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.extension.ExtendWith;
@@ -12,6 +12,7 @@ import org.mockito.junit.jupiter.MockitoExtension;
import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
@@ -24,15 +25,14 @@ class ZiniaoShopSwitchServiceTest {
@Mock @Mock
private ZiniaoShopIndexService ziniaoShopIndexService; private ZiniaoShopIndexService ziniaoShopIndexService;
@Mock @Mock
private ShopManageService shopManageService; private ManagedShopNamePort managedShopNamePort;
@Test @Test
void missingManagedShopStopsIndexedLookup() { void missingManagedShopStopsIndexedLookup() {
String message = "后台店铺管理中未找到店铺:缺失店铺,请先添加店铺信息"; String message = "后台店铺管理中未找到店铺:缺失店铺,请先添加店铺信息";
when(shopManageService.requireShopByName("缺失店铺")) doThrow(new BusinessException(message)).when(managedShopNamePort).requireShopByName("缺失店铺");
.thenThrow(new BusinessException(message));
ZiniaoShopSwitchService service = new ZiniaoShopSwitchService( ZiniaoShopSwitchService service = new ZiniaoShopSwitchService(
ziniaoAuthService, ziniaoShopIndexService, shopManageService); ziniaoAuthService, ziniaoShopIndexService, managedShopNamePort);
BusinessException exception = assertThrows(BusinessException.class, BusinessException exception = assertThrows(BusinessException.class,
() -> service.findIndexedStoreByName("缺失店铺", false)); () -> service.findIndexedStoreByName("缺失店铺", false));
@@ -47,22 +47,22 @@ class ZiniaoShopSwitchServiceTest {
expected.setMatched(true); expected.setMatched(true);
when(ziniaoShopIndexService.findIndexedStoreByName("测试店铺", false)).thenReturn(expected); when(ziniaoShopIndexService.findIndexedStoreByName("测试店铺", false)).thenReturn(expected);
ZiniaoShopSwitchService service = new ZiniaoShopSwitchService( ZiniaoShopSwitchService service = new ZiniaoShopSwitchService(
ziniaoAuthService, ziniaoShopIndexService, shopManageService); ziniaoAuthService, ziniaoShopIndexService, managedShopNamePort);
ZiniaoShopMatchResultVo actual = service.findIndexedStoreByName(" 测试店铺 ", false); ZiniaoShopMatchResultVo actual = service.findIndexedStoreByName(" 测试店铺 ", false);
assertSame(expected, actual); assertSame(expected, actual);
InOrder order = inOrder(shopManageService, ziniaoShopIndexService); InOrder order = inOrder(managedShopNamePort, ziniaoShopIndexService);
order.verify(shopManageService).requireShopByName("测试店铺"); order.verify(managedShopNamePort).requireShopByName("测试店铺");
order.verify(ziniaoShopIndexService).findIndexedStoreByName("测试店铺", false); order.verify(ziniaoShopIndexService).findIndexedStoreByName("测试店铺", false);
} }
@Test @Test
void missingManagedShopStopsDirectStaffMatch() { void missingManagedShopStopsDirectStaffMatch() {
when(shopManageService.requireShopByName("缺失店铺")) doThrow(new BusinessException("后台店铺管理中未找到店铺:缺失店铺,请先添加店铺信息"))
.thenThrow(new BusinessException("后台店铺管理中未找到店铺:缺失店铺,请先添加店铺信息")); .when(managedShopNamePort).requireShopByName("缺失店铺");
ZiniaoShopSwitchService service = new ZiniaoShopSwitchService( ZiniaoShopSwitchService service = new ZiniaoShopSwitchService(
ziniaoAuthService, ziniaoShopIndexService, shopManageService); ziniaoAuthService, ziniaoShopIndexService, managedShopNamePort);
assertThrows(BusinessException.class, assertThrows(BusinessException.class,
() -> service.matchStoreByNameAcrossStaff("缺失店铺", 12L)); () -> service.matchStoreByNameAcrossStaff("缺失店铺", 12L));
@@ -72,12 +72,12 @@ class ZiniaoShopSwitchServiceTest {
@Test @Test
void blankShopNameDoesNotQueryManagedShopOrZiniao() { void blankShopNameDoesNotQueryManagedShopOrZiniao() {
ZiniaoShopSwitchService service = new ZiniaoShopSwitchService( ZiniaoShopSwitchService service = new ZiniaoShopSwitchService(
ziniaoAuthService, ziniaoShopIndexService, shopManageService); ziniaoAuthService, ziniaoShopIndexService, managedShopNamePort);
ZiniaoShopMatchResultVo result = service.findIndexedStoreByName("  ", false); ZiniaoShopMatchResultVo result = service.findIndexedStoreByName("  ", false);
assertEquals(ZiniaoShopIndexService.MATCH_STATUS_PENDING, result.getMatchStatus()); assertEquals(ZiniaoShopIndexService.MATCH_STATUS_PENDING, result.getMatchStatus());
assertEquals("店铺名为空,无法匹配索引", result.getMatchMessage()); assertEquals("店铺名为空,无法匹配索引", result.getMatchMessage());
verifyNoInteractions(shopManageService, ziniaoShopIndexService, ziniaoAuthService); verifyNoInteractions(managedShopNamePort, ziniaoShopIndexService, ziniaoAuthService);
} }
} }