feat(保留期清理): 补齐 7 处只增不删的数据;修结果对象孤儿

审核发现一批表/元数据只增不删,且对象存储存在活跃泄漏:

- module-cleanup 漏收 task_result_item / task_result_payload 的 payload 指针,
  每晚删完行就在 json-server 桶留下孤儿对象;指针收集触顶从"截断后照样删整组"
  改为任务组二分拆分,拆到单任务仍触顶则整组不删并记 error(截断会让指针随行消失)
- 结果文件对象此前从不回收(只删 DB 行、对象留在桶里),现改为事务提交后
  按 file_result / task_file_job 的 result_file_url 逐个回收
- 新增 7 个保留期清理:device_log_file 元数据、price_track_loop_run、撞款扫描行、
  密钥用量日统计、紫鸟记忆过期行、PUBLISH 任务、BRAND 任务。后两者刻意不复用
  module-cleanup(BRAND 不写 biz_file_task;PUBLISH 单任务可达数万行),
  改为调各自既有的业务删除入口,保证结果对象与子表一起回收
- 未读通知补保留期(180 天,已读仍 90 天);ZiniaoMemoryStoreService.deleteExpired
  此前全库无调用方(实现了没接线),现已挂定时任务

全部沿用既有范式:@Scheduled + 分布式锁单实例 + 分批 + 单轮批次数上限 + 失败只记日志。
This commit is contained in:
2026-09-16 01:59:28 +08:00
parent 5e5816cd74
commit f2ada02383
29 changed files with 1680 additions and 32 deletions
@@ -67,4 +67,12 @@ public class NotificationProperties {
/** 已读通知保留天数(超期自动清理),默认 90 天。 */
private int readRetentionDays = 90;
/**
* 未读通知保留天数,默认 180 天(比已读长一倍)。
*
* <p>未读通知此前永不清理:不看铃铛的用户会无限累积。保留期给得更宽,
* 是因为未读意味着"用户可能还没看到",但也不能永远留着。
*/
private int unreadRetentionDays = 180;
}
@@ -3,7 +3,31 @@ package com.nanri.aiimage.modules.brand.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.nanri.aiimage.modules.brand.model.entity.BrandCrawlTaskEntity;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import java.time.LocalDateTime;
import java.util.List;
@Mapper
public interface BrandCrawlTaskMapper extends BaseMapper<BrandCrawlTaskEntity> {
/**
* 查询超过保留期的终态品牌检测任务 id(保留期清理用,只取 id 不拉整行——历史行的
* file_paths/result_paths JSON 字段可能很大)。
*
* <p>终态集合与 {@code BrandTaskService} 的状态机一致(success/failed/cancelled),
* pending/running 绝不返回(删了正在跑的任务,结果回传会找不到任务行)。
* 时间线用 updated_at,与 BrandTaskStaleRepairSpiImpl 的陈旧判定同款口径,
* 可命中 V120 的 idx_brand_crawl_task_status_updated(status, updated_at) 索引。
*/
@Select("""
SELECT id FROM brand_crawl_tasks
WHERE status IN ('success', 'failed', 'cancelled')
AND updated_at < #{cutoff}
ORDER BY id ASC
LIMIT #{batchSize}
""")
List<Long> selectExpiredTerminalTaskIds(@Param("cutoff") LocalDateTime cutoff,
@Param("batchSize") int batchSize);
}
@@ -0,0 +1,93 @@
package com.nanri.aiimage.modules.brand.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.brand.mapper.BrandCrawlTaskMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
/**
* 品牌检测任务(brand_crawl_tasks)的保留期清理(2026-09 审核:该表只增不删,永久累积)。
*
* <p>该表自建、不写 biz_file_task,故不在 ModuleHistoryCleanupService 的清理名单里,
* 此前没有任何删除路径。这里只查过期终态任务的 id,逐个走
* {@link BrandTaskService#deleteTask(Long)} 既有删除入口 —— 它已处理任务行删除 +
* 存储数据清理(brandTaskStorageService.deleteTaskData+ 进度缓存清理,本类不重新实现删除逻辑。
*
* <p>双节点用 job 锁保证单实例执行;每批小批量(默认 50)逐个删,避免单次跑太久占住锁。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class BrandTaskRetentionService {
/** 单轮最多处理的批次数(每批 batchSize 个任务),剩余留给下一轮。 */
static final int MAX_BATCHES_PER_RUN = 20;
private final BrandCrawlTaskMapper brandCrawlTaskMapper;
private final BrandTaskService brandTaskService;
private final DistributedJobLockService distributedJobLockService;
@Value("${aiimage.brand.task-retention-days:90}")
private int retentionDays = 90;
@Value("${aiimage.brand.task-retention-batch-size:50}")
private int retentionBatchSize = 50;
@Scheduled(cron = "${aiimage.brand.task-retention-cron:0 45 4 * * *}")
public void purgeExpiredTasks() {
int days = Math.max(1, retentionDays);
int batchSize = Math.max(1, retentionBatchSize);
LocalDateTime cutoff = LocalDateTime.now().minusDays(days);
// 单轮 20 批 × 50 个任务的删除(含存储数据清理)可能跑较久,锁 TTL 给足 30 分钟
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("brand:task-retention", Duration.ofMinutes(30));
if (lockHandle == null) {
log.info("[brand-retention] 任务保留清理跳过:另一实例持锁");
return;
}
try (lockHandle) {
int totalDeleted = 0;
int totalFailed = 0;
int batches = 0;
while (batches < MAX_BATCHES_PER_RUN) {
// 只取 id:历史行的 file_paths/result_paths JSON 字段可能很大
List<Long> taskIds = brandCrawlTaskMapper.selectExpiredTerminalTaskIds(cutoff, batchSize);
if (taskIds.isEmpty()) {
break;
}
batches++;
int deletedInBatch = 0;
for (Long taskId : taskIds) {
try {
brandTaskService.deleteTask(taskId);
deletedInBatch++;
} catch (Exception ex) {
// 单个任务删除失败只记日志继续:一个坏任务不能卡住整轮
log.warn("[brand-retention] 任务删除失败 taskId={} msg={}", taskId, ex.getMessage(), ex);
}
}
totalDeleted += deletedInBatch;
totalFailed += taskIds.size() - deletedInBatch;
if (deletedInBatch == 0) {
log.warn("[brand-retention] 整批任务均删除失败,提前结束本轮以免反复重试同一批");
break;
}
if (taskIds.size() < batchSize) {
break;
}
}
log.info("[brand-retention] 任务保留清理完成 cutoff={} retentionDays={} deleted={} failed={} batches={}",
cutoff, days, totalDeleted, totalFailed, batches);
} catch (Exception ex) {
log.warn("[brand-retention] 任务保留清理失败 cutoff={} msg={}", cutoff, ex.getMessage(), ex);
}
}
}
@@ -2,8 +2,25 @@ package com.nanri.aiimage.modules.devicelog.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.nanri.aiimage.modules.devicelog.model.entity.DeviceLogFileEntity;
import org.apache.ibatis.annotations.Delete;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import java.time.LocalDate;
@Mapper
public interface DeviceLogFileMapper extends BaseMapper<DeviceLogFileEntity> {
/**
* 按日志日期分批删除保留期外的元数据行。
*
* <p>只删 {@code log_date} 早于 cutoff 的行:这条线正好是查询侧可见窗口的边界,
* 即"早已看不见、只剩占位"的行,删掉不影响任何读取路径。
*/
@Delete("""
DELETE FROM device_log_file
WHERE log_date < #{cutoff}
LIMIT #{batchSize}
""")
int deleteOlderThanBatch(@Param("cutoff") LocalDate cutoff, @Param("batchSize") int batchSize);
}
@@ -0,0 +1,77 @@
package com.nanri.aiimage.modules.devicelog.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.DeviceLogOssProperties;
import com.nanri.aiimage.modules.devicelog.mapper.DeviceLogFileMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDate;
/**
* 设备日志元数据的保留期清理。
*
* <p>对象侧一直有生命周期(主机B 独立 MinIO 桶,由运维 mc 定时任务按 7 天回收),
* 但 {@code device_log_file} 的元数据行此前**只增不删**:客户端每 60 秒上报一轮,
* 每台设备每天登记若干「来源+设备+文件名+日期」行,长期运行会无限累积。
*
* <p>查询侧本就只展示保留期内的行({@link DeviceLogService#page} 用同一个
* {@code retentionDays} 过滤),这里按完全相同的边界删掉早已不可见的行。
* 保留天数直接复用 {@code aiimage.device-log-oss.retention-days}
* 与对象侧共用同一个配置项,避免两边各写一份天数后悄悄漂移。
*
* <p>对象本身仍由桶生命周期负责回收,本任务不碰对象:两条时间线按「对象修改时间」
* 与「日志日期」衡量,可能有几天错位,但对象最终仍会被桶规则删除,不会永久残留。
*
* <p>双节点用 job 锁保证单实例执行;分批删除并限制单轮批次数,避免一次跑太久占住锁。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class DeviceLogRetentionService {
/** 单轮最多删除的批次数(每批 batchSize 行),剩余留给下一轮。 */
static final int MAX_BATCHES_PER_RUN = 20;
private final DeviceLogFileMapper deviceLogFileMapper;
private final DeviceLogOssProperties properties;
private final DistributedJobLockService distributedJobLockService;
@Value("${aiimage.device-log.retention-batch-size:2000}")
private int retentionBatchSize = 2000;
@Scheduled(cron = "${aiimage.device-log.retention-cron:0 40 4 * * *}")
public void purgeExpiredMetadata() {
int days = properties.retentionDaysOrDefault();
int batchSize = Math.max(100, retentionBatchSize);
// 与查询侧可见窗口同一条边界:page() 只展示 log_date >= 今天-(days-1) 的行
LocalDate cutoff = LocalDate.now().minusDays(days - 1L);
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("device-log:metadata-retention", Duration.ofMinutes(15));
if (lockHandle == null) {
log.info("[device-log] 元数据保留清理跳过:另一实例持锁");
return;
}
try (lockHandle) {
int totalDeleted = 0;
int batches = 0;
while (batches < MAX_BATCHES_PER_RUN) {
int deleted = deviceLogFileMapper.deleteOlderThanBatch(cutoff, batchSize);
batches++;
totalDeleted += deleted;
if (deleted < batchSize) {
break;
}
}
log.info("[device-log] 元数据保留清理完成 cutoff={} retentionDays={} deleted={} batches={}",
cutoff, days, totalDeleted, batches);
} catch (Exception ex) {
log.warn("[device-log] 元数据保留清理失败 cutoff={} msg={}", cutoff, ex.getMessage(), ex);
}
}
}
@@ -273,15 +273,17 @@ public class NotificationScanScheduler {
"service_down:" + serviceKey + ":" + hour, null);
}
/** 已读通知保留期清理:每天最多一次。 */
/** 通知保留期清理:每天最多一次;已读与未读各按自己的保留天数。 */
private void cleanupExpiredIfNeeded() {
LocalDate today = LocalDate.now();
if (today.equals(lastCleanupDate)) {
return;
}
lastCleanupDate = today;
LocalDateTime cutoff = LocalDateTime.now().minusDays(Math.max(1, properties.getReadRetentionDays()));
notificationService.cleanupReadBefore(cutoff);
LocalDateTime readCutoff = LocalDateTime.now().minusDays(Math.max(1, properties.getReadRetentionDays()));
notificationService.cleanupReadBefore(readCutoff);
LocalDateTime unreadCutoff = LocalDateTime.now().minusDays(Math.max(1, properties.getUnreadRetentionDays()));
notificationService.cleanupUnreadBefore(unreadCutoff);
}
private String dedupeKeyOf(BucketKey key) {
@@ -217,6 +217,17 @@ public class NotificationService {
return deleted;
}
/** 清理指定时间之前仍未读的通知(保留期由调用方决定,比已读给得更宽)。 */
public int cleanupUnreadBefore(LocalDateTime cutoff) {
int deleted = userNotificationMapper.delete(new LambdaQueryWrapper<UserNotificationEntity>()
.isNull(UserNotificationEntity::getReadAt)
.lt(UserNotificationEntity::getCreatedAt, cutoff));
if (deleted > 0) {
log.info("[notification] 清理历史未读通知 cutoff={} 删除={} 条", cutoff, deleted);
}
return deleted;
}
private LambdaQueryWrapper<UserNotificationEntity> baseWrapper(Long userId, String audience, boolean onlyUnread,
NotificationPageQuery query) {
LambdaQueryWrapper<UserNotificationEntity> wrapper = new LambdaQueryWrapper<UserNotificationEntity>()
@@ -2,8 +2,26 @@ package com.nanri.aiimage.modules.pricetrack.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.nanri.aiimage.modules.pricetrack.model.entity.PriceTrackLoopRunEntity;
import org.apache.ibatis.annotations.Delete;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import java.time.LocalDateTime;
@Mapper
public interface PriceTrackLoopRunMapper extends BaseMapper<PriceTrackLoopRunEntity> {
}
/**
* 分批删除已结束且超过保留期的循环批次行。
*
* <p>只删终态行:RUNNING 的批次删了会让后续 dispatch / childFinished 找不到记录。
* 时间线用 COALESCE 兜底——终态行本应写 finished_at,历史行可能只有 updated_at。
*/
@Delete("""
DELETE FROM biz_price_track_loop_run
WHERE status IN ('SUCCESS', 'FAILED', 'STOPPED')
AND COALESCE(finished_at, updated_at, created_at) < #{cutoff}
LIMIT #{batchSize}
""")
int deleteFinishedBefore(@Param("cutoff") LocalDateTime cutoff, @Param("batchSize") int batchSize);
}
@@ -0,0 +1,70 @@
package com.nanri.aiimage.modules.pricetrack.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.pricetrack.mapper.PriceTrackLoopRunMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDateTime;
/**
* 跟价循环批次表(biz_price_track_loop_run)的保留期清理。
*
* <p>每次「循环跟价」插一行,且只在请求停止时更新——全模块此前没有任何删除路径(纯遗漏)。
* 读取侧只有两个入口:按 id 取单条(用户当下正在看的那一轮)与「当前是否有活跃循环」,
* 没有任何历史列表查询,所以终态批次过了保留期即可安全删除。
*
* <p>只删终态(SUCCESS/FAILED/STOPPED)行,RUNNING 一律不动——正在跑的循环被删会让
* 后续 dispatch / childFinished 找不到记录而中断。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class PriceTrackLoopRunRetentionService {
/** 单轮最多删除的批次数,剩余留给下一轮。 */
static final int MAX_BATCHES_PER_RUN = 20;
private final PriceTrackLoopRunMapper loopRunMapper;
private final DistributedJobLockService distributedJobLockService;
@Value("${aiimage.price-track.loop-run-retention-days:30}")
private int retentionDays = 30;
@Value("${aiimage.price-track.loop-run-retention-batch-size:500}")
private int retentionBatchSize = 500;
@Scheduled(cron = "${aiimage.price-track.loop-run-retention-cron:0 20 4 * * *}")
public void purgeExpiredLoopRuns() {
int days = Math.max(1, retentionDays);
int batchSize = Math.max(50, retentionBatchSize);
LocalDateTime cutoff = LocalDateTime.now().minusDays(days);
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("price-track:loop-run-retention", Duration.ofMinutes(15));
if (lockHandle == null) {
log.info("[price-track-loop] 批次保留清理跳过:另一实例持锁");
return;
}
try (lockHandle) {
int totalDeleted = 0;
int batches = 0;
while (batches < MAX_BATCHES_PER_RUN) {
int deleted = loopRunMapper.deleteFinishedBefore(cutoff, batchSize);
batches++;
totalDeleted += deleted;
if (deleted < batchSize) {
break;
}
}
log.info("[price-track-loop] 批次保留清理完成 cutoff={} retentionDays={} deleted={} batches={}",
cutoff, days, totalDeleted, batches);
} catch (Exception ex) {
log.warn("[price-track-loop] 批次保留清理失败 cutoff={} msg={}", cutoff, ex.getMessage(), ex);
}
}
}
@@ -0,0 +1,97 @@
package com.nanri.aiimage.modules.publish.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
/**
* 上架任务的保留期清理(2026-09 审核:PUBLISH 的任务历史此前完全没有清理,永久累积)。
*
* <p>涉及 biz_file_task / biz_file_result / biz_publish_item(每个上传 Excel 的每一行落一行,
* 单任务可达数万行)/ biz_publish_file,此前只有用户手动删任务才回收。这里刻意不复用
* ModuleHistoryCleanupService:上架单任务行数太大,且删除必须经过业务删除入口回收结果对象。
*
* <p>删除动作复用 {@link PublishTaskService#deleteTaskForRetention(Long)}(与用户删任务
* 同一套删除实现:明细/结果/分片载荷 + 事务提交后回收结果对象),本类只做「查一批过期 id →
* 逐个调用 → 计数」。该入口保留状态校验,只删终态任务,PENDING/RUNNING 不会被碰。
*
* <p>双节点用 job 锁保证单实例执行;每批小批量(默认 50)逐个删,避免单次事务过长。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class PublishTaskRetentionService {
/** 单轮最多处理的批次数(每批 batchSize 个任务),剩余留给下一轮。 */
static final int MAX_BATCHES_PER_RUN = 20;
private final FileTaskMapper fileTaskMapper;
private final PublishTaskService publishTaskService;
private final DistributedJobLockService distributedJobLockService;
@Value("${aiimage.publish.task-retention-days:90}")
private int retentionDays = 90;
@Value("${aiimage.publish.task-retention-batch-size:50}")
private int retentionBatchSize = 50;
@Scheduled(cron = "${aiimage.publish.task-retention-cron:0 30 4 * * *}")
public void purgeExpiredTasks() {
int days = Math.max(1, retentionDays);
int batchSize = Math.max(1, retentionBatchSize);
LocalDateTime cutoff = LocalDateTime.now().minusDays(days);
// 单轮 20 批 × 50 个任务的域内删除(含结果对象回收)可能跑较久,锁 TTL 给足 30 分钟
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("publish:task-retention", Duration.ofMinutes(30));
if (lockHandle == null) {
log.info("[publish-retention] 任务保留清理跳过:另一实例持锁");
return;
}
try (lockHandle) {
int totalDeleted = 0;
int totalFailed = 0;
int batches = 0;
while (batches < MAX_BATCHES_PER_RUN) {
// 只取 id:上架任务行可携带很大的 request_json/result_json,不拉整行
List<Long> taskIds = fileTaskMapper.selectExpiredTerminalTaskIds(
PublishTaskService.MODULE_TYPE, cutoff, batchSize);
if (taskIds.isEmpty()) {
break;
}
batches++;
int deletedInBatch = 0;
for (Long taskId : taskIds) {
try {
publishTaskService.deleteTaskForRetention(taskId);
deletedInBatch++;
} catch (Exception ex) {
// 单个任务删除失败只记日志继续:一个坏任务不能卡住整轮
log.warn("[publish-retention] 任务删除失败 taskId={} msg={}", taskId, ex.getMessage(), ex);
}
}
totalDeleted += deletedInBatch;
totalFailed += taskIds.size() - deletedInBatch;
if (deletedInBatch == 0) {
log.warn("[publish-retention] 整批任务均删除失败,提前结束本轮以免反复重试同一批");
break;
}
if (taskIds.size() < batchSize) {
break;
}
}
log.info("[publish-retention] 任务保留清理完成 cutoff={} retentionDays={} deleted={} failed={} batches={}",
cutoff, days, totalDeleted, totalFailed, batches);
} catch (Exception ex) {
log.warn("[publish-retention] 任务保留清理失败 cutoff={} msg={}", cutoff, ex.getMessage(), ex);
}
}
}
@@ -96,6 +96,12 @@ public class PublishTaskService {
private static final String STATUS_RUNNING = "RUNNING";
private static final String STATUS_SUCCESS = "SUCCESS";
private static final String STATUS_FAILED = "FAILED";
/**
* 保留期清理允许删除的终态集合(与 ModuleHistoryCleanupService 的 TERMINAL_STATUSES 一致)。
* PENDING/RUNNING 绝不在此列;PUBLISH 实际只会写 SUCCESS/FAILED,取消态是无害的超集。
*/
private static final Set<String> RETENTION_TERMINAL_STATUSES =
Set.of(STATUS_SUCCESS, STATUS_FAILED, "CANCELLED", "CANCELED");
private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private final LocalFileStorageService localFileStorageService;
@@ -566,6 +572,37 @@ public class PublishTaskService {
@Transactional
public void deleteTask(Long taskId, Long userId) {
FileTaskEntity task = requireTaskForDeletion(taskId, userId);
deleteTaskWithRelatedRows(task);
}
/**
* 保留期清理专用删除入口(内部调用,定时任务没有用户身份)。
*
* <p>与 {@link #deleteTask(Long, Long)} 复用同一套删除实现(明细行/结果记录/分片载荷 +
* 事务提交后回收结果对象),仅跳过用户归属校验;状态校验保留 —— 只允许删终态任务,
* PENDING/RUNNING 一律不删(正在跑的任务被删会让分片回传、结果组装找不到任务行)。
* 行已被删除或非上架任务时幂等跳过,不抛异常。
*/
@Transactional
public void deleteTaskForRetention(Long taskId) {
if (taskId == null || taskId <= 0) {
throw new BusinessException("taskId 不合法");
}
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
log.info("[publish] 保留期删除跳过:任务已不存在或非上架任务 taskId={}", taskId);
return;
}
if (!isRetentionTerminal(task.getStatus())) {
log.warn("[publish] 保留期删除跳过非终态任务 taskId={} status={}", taskId, task.getStatus());
return;
}
deleteTaskWithRelatedRows(task);
}
/** 删除任务行与全部关联数据,并在事务提交后回收远端对象(deleteTask 与保留期清理共用)。 */
private void deleteTaskWithRelatedRows(FileTaskEntity task) {
Long taskId = task.getId();
List<FileResultEntity> results = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getTaskId, taskId)
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
@@ -1953,6 +1990,11 @@ public class PublishTaskService {
return STATUS_SUCCESS.equals(status) || STATUS_FAILED.equals(status);
}
/** 保留期清理的终态判定:大小写不敏感,PENDING/RUNNING 一律返回 false。 */
private static boolean isRetentionTerminal(String status) {
return status != null && RETENTION_TERMINAL_STATUSES.contains(status.trim().toUpperCase(Locale.ROOT));
}
private List<Long> normalizeTaskIds(List<Long> taskIds) {
if (taskIds == null) {
return List.of();
@@ -4,9 +4,13 @@ import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ScanLightRowDto;
import com.nanri.aiimage.modules.shopduplicatecheck.model.dto.ScanFullRowDto;
import com.nanri.aiimage.modules.shopduplicatecheck.model.entity.ShopDataDuplicateScanEntity;
import org.apache.ibatis.annotations.Delete;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import java.time.LocalDateTime;
@Mapper
public interface ShopDataDuplicateScanMapper extends BaseMapper<ShopDataDuplicateScanEntity> {
@@ -20,4 +24,19 @@ public interface ShopDataDuplicateScanMapper extends BaseMapper<ShopDataDuplicat
+ "created_at AS createdAt FROM shop_data_duplicate_scan "
+ "WHERE status = 'SUCCESS' ORDER BY id DESC LIMIT 1")
ScanFullRowDto selectLatestFullRow();
/** 第 N 新的行 id(offset 从 0 计);行数不足时返回 null,供保留清理计算保护线。 */
@Select("SELECT id FROM shop_data_duplicate_scan ORDER BY id DESC LIMIT 1 OFFSET #{offset}")
Long selectNthNewestId(@Param("offset") int offset);
/** 分批删除保留期外、且早于保护线的历史扫描行。 */
@Delete("""
DELETE FROM shop_data_duplicate_scan
WHERE created_at < #{cutoff}
AND id < #{protectFromId}
LIMIT #{batchSize}
""")
int deleteOlderThanBatch(@Param("cutoff") LocalDateTime cutoff,
@Param("protectFromId") long protectFromId,
@Param("batchSize") int batchSize);
}
@@ -0,0 +1,76 @@
package com.nanri.aiimage.modules.shopduplicatecheck.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDataDuplicateScanMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDateTime;
/**
* 撞款扫描结果表(shop_data_duplicate_scan)的保留期清理。
*
* <p>扫描每晚自动跑一轮、管理员还能手动触发,每次插入一行**含整份聚合 payload 的结果**
* 但读取侧只认最新一行({@code selectLatestLightRow} / {@code selectLatestFullRow} 都是
* {@code ORDER BY id DESC LIMIT 1}),历史行纯属占用磁盘,此前无任何删除路径。
*
* <p>保护策略:无论多旧,始终保留按 id 最新的若干行——
* 万一扫描停摆很久,界面上仍能看到最后一份结果,而不是被清理任务顺手抹掉。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class ShopDataDuplicateScanRetentionService {
/** 单轮最多删除的批次数,剩余留给下一轮。 */
static final int MAX_BATCHES_PER_RUN = 20;
/** 无论时间多久都保留的最新行数(最新 SUCCESS 行必然在其中)。 */
static final int PROTECTED_ROWS = 10;
private final ShopDataDuplicateScanMapper scanMapper;
private final DistributedJobLockService distributedJobLockService;
@Value("${aiimage.shop-duplicate-check.scan-retention-days:90}")
private int retentionDays = 90;
@Value("${aiimage.shop-duplicate-check.scan-retention-batch-size:200}")
private int retentionBatchSize = 200;
@Scheduled(cron = "${aiimage.shop-duplicate-check.scan-retention-cron:0 50 3 * * *}")
public void purgeExpiredScans() {
int days = Math.max(1, retentionDays);
int batchSize = Math.max(20, retentionBatchSize);
LocalDateTime cutoff = LocalDateTime.now().minusDays(days);
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("shop-duplicate-check:scan-retention", Duration.ofMinutes(15));
if (lockHandle == null) {
log.info("[shop-duplicate-check] 扫描结果保留清理跳过:另一实例持锁");
return;
}
try (lockHandle) {
Long nthNewestId = scanMapper.selectNthNewestId(PROTECTED_ROWS - 1);
// 行数不足保护量时没什么可删,直接返回
long protectFromId = nthNewestId == null ? Long.MAX_VALUE : nthNewestId;
int totalDeleted = 0;
int batches = 0;
while (batches < MAX_BATCHES_PER_RUN) {
int deleted = scanMapper.deleteOlderThanBatch(cutoff, protectFromId, batchSize);
batches++;
totalDeleted += deleted;
if (deleted < batchSize) {
break;
}
}
log.info("[shop-duplicate-check] 扫描结果保留清理完成 cutoff={} retentionDays={} protectFromId={} deleted={} batches={}",
cutoff, days, protectFromId, totalDeleted, batches);
} catch (Exception ex) {
log.warn("[shop-duplicate-check] 扫描结果保留清理失败 cutoff={} msg={}", cutoff, ex.getMessage(), ex);
}
}
}
@@ -3,7 +3,35 @@ package com.nanri.aiimage.modules.task.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
import org.apache.ibatis.annotations.Select;
import java.time.LocalDateTime;
import java.util.List;
@Mapper
public interface FileTaskMapper extends BaseMapper<FileTaskEntity> {
/**
* 查询某模块下超过保留期的终态任务 id(保留期清理用,只取 id 不拉整行)。
*
* <p>终态集合与 ModuleHistoryCleanupService 的 TERMINAL_STATUSES 一致:PENDING/RUNNING 绝不返回——
* 正在跑的任务被删会让分片回传、结果组装找不到任务行。
*
* <p>时间线只用 updated_at:该列由 DB 的 ON UPDATE CURRENT_TIMESTAMP 维护,任何一次行更新
* (含终态写入)都会刷新,不会早于 finished_atfinished_at 在任务重置时会被置空,不可单独依赖);
* 且能命中 V120 的 idx_biz_file_task_status_updated(status, updated_at) 索引,避免每轮全表扫描。
* 注意不要在这里包 COALESCE(finished_at, updated_at):表达式会让该索引失效。
*/
@Select("""
SELECT id FROM biz_file_task
WHERE module_type = #{moduleType}
AND status IN ('SUCCESS', 'FAILED', 'CANCELLED', 'CANCELED')
AND updated_at < #{cutoff}
ORDER BY id ASC
LIMIT #{batchSize}
""")
List<Long> selectExpiredTerminalTaskIds(@Param("moduleType") String moduleType,
@Param("cutoff") LocalDateTime cutoff,
@Param("batchSize") int batchSize);
}
@@ -4,6 +4,7 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.ModuleCleanupProperties;
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
import com.nanri.aiimage.modules.task.spi.CollectDataItemCleanupSpi;
import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
@@ -44,6 +45,8 @@ public class ModuleHistoryCleanupService {
private static final String COLLECT_DATA_MODULE_TYPE = "COLLECT_DATA";
private static final int DEFAULT_BATCH_SIZE = 500;
private static final int LOG_SAMPLE_IDS = 5;
/** 指针收集触顶时任务组的最大二分拆分深度(500 任务拆到单个任务约需 9 层)。 */
private static final int MAX_POINTER_SPLIT_DEPTH = 20;
private final ModuleCleanupProperties moduleCleanupProperties;
private final FileTaskMapper fileTaskMapper;
@@ -57,14 +60,15 @@ public class ModuleHistoryCleanupService {
private final CollectDataItemCleanupSpi collectDataItemCleanupSpi;
private final DistributedJobLockService distributedJobLockService;
private final TransientPayloadDeleteOrchestrator transientPayloadDeleteOrchestrator;
private final OssStorageService ossStorageService;
private final TransactionTemplate transactionTemplate;
/**
* 单最多收集的 payload 指针数:超过即截断(保底可重试),
* 单最多收集的 payload 指针数:超过即把任务组二分拆分(见 {@link #cleanupTaskGroup}),
* 防止单任务行数异常巨大时无界收集造成内存增长。
*/
@Value("${aiimage.module-cleanup.max-collect-payloads:10000}")
private int maxCollectPayloadsPerRun = 10000;
@Value("${aiimage.module-cleanup.max-collect-payloads:50000}")
private int maxCollectPayloadsPerRun = 50000;
public ModuleHistoryCleanupService(ModuleCleanupProperties moduleCleanupProperties,
FileTaskMapper fileTaskMapper,
@@ -78,6 +82,7 @@ public class ModuleHistoryCleanupService {
CollectDataItemCleanupSpi collectDataItemCleanupSpi,
DistributedJobLockService distributedJobLockService,
TransientPayloadDeleteOrchestrator transientPayloadDeleteOrchestrator,
OssStorageService ossStorageService,
PlatformTransactionManager platformTransactionManager) {
this.moduleCleanupProperties = moduleCleanupProperties;
this.fileTaskMapper = fileTaskMapper;
@@ -91,6 +96,7 @@ public class ModuleHistoryCleanupService {
this.collectDataItemCleanupSpi = collectDataItemCleanupSpi;
this.distributedJobLockService = distributedJobLockService;
this.transientPayloadDeleteOrchestrator = transientPayloadDeleteOrchestrator;
this.ossStorageService = ossStorageService;
this.transactionTemplate = new TransactionTemplate(platformTransactionManager);
}
@@ -161,20 +167,11 @@ public class ModuleHistoryCleanupService {
}
}
if (!batchTaskIds.isEmpty()) {
final List<Long> taskIds = batchTaskIds;
final List<Long> collectDataTaskIds = batchCollectDataTaskIds;
final List<String> types = moduleTypes;
final List<String> collected = new ArrayList<>();
transactionTemplate.executeWithoutResult(status -> {
collected.addAll(collectPayloadPointers(types, taskIds, maxCollectPayloadsPerRun));
int deletedRows = deleteRows(types, taskIds, collectDataTaskIds);
submitAndFlush(collected);
log.info("[module-cleanup] batch: taskIds={}, deletedRows={}, collectedPointers={}",
formatIdSample(taskIds, LOG_SAMPLE_IDS), deletedRows, collected.size());
});
totalBatches++;
totalDeletedTasks += taskIds.size();
totalCollectedPointers += collected.size();
BatchTally tally = new BatchTally();
cleanupTaskGroup(moduleTypes, batchTaskIds, batchCollectDataTaskIds, 0, tally);
totalBatches += tally.batches;
totalDeletedTasks += tally.deletedTasks;
totalCollectedPointers += tally.collectedPointers;
}
if (pageMaxId <= cursor) {
log.warn("[module-cleanup] keyset cursor did not advance, abort loop cursor={}", cursor);
@@ -190,7 +187,107 @@ public class ModuleHistoryCleanupService {
}
}
/** 一个事务内完成:收集指针 → 删除本批行 → 行删完后提交清理队列并 flush。 */
/**
* 清理一组过期任务:收集 payload 指针 → 删除行 → 提交对象回收。
*
* <p>指针数触顶时把任务组二分拆分重试,保证「凡被删除的行,其 transient 指针一定已进回收队列」——
* 截断会让指针随行一起消失,对象再无引用可查、永久留在桶里。
* 拆到单个任务仍触顶则整组不删并记 error(宁可留脏行待下轮重试,也不制造孤儿对象);
* 运维可据此调大 {@code aiimage.module-cleanup.max-collect-payloads} 后自动收敛。
*/
private void cleanupTaskGroup(List<String> moduleTypes, List<Long> taskIds, List<Long> collectDataTaskIds,
int depth, BatchTally tally) {
if (taskIds.isEmpty()) {
return;
}
List<String> collected = collectPayloadPointers(moduleTypes, taskIds, maxCollectPayloadsPerRun);
boolean capped = collected.size() >= maxCollectPayloadsPerRun;
if (capped && taskIds.size() > 1 && depth < MAX_POINTER_SPLIT_DEPTH) {
int half = taskIds.size() / 2;
log.warn("[module-cleanup] payload 指针数触顶 {},任务组二分拆分重试: depth={}, size={}",
maxCollectPayloadsPerRun, depth, taskIds.size());
cleanupTaskGroup(moduleTypes, new ArrayList<>(taskIds.subList(0, half)),
collectDataTaskIds, depth + 1, tally);
cleanupTaskGroup(moduleTypes, new ArrayList<>(taskIds.subList(half, taskIds.size())),
collectDataTaskIds, depth + 1, tally);
return;
}
if (capped) {
log.error("[module-cleanup] payload 指针数触顶且无法再拆分,本轮保留任务行待下轮重试: taskIds={}",
formatIdSample(taskIds, LOG_SAMPLE_IDS));
return;
}
final List<Long> ids = List.copyOf(taskIds);
final Set<Long> collectDataIdSet = Set.copyOf(collectDataTaskIds);
final List<Long> groupCollectDataIds = ids.stream().filter(collectDataIdSet::contains).toList();
List<String> resultObjectKeys = collectResultObjectKeys(moduleTypes, ids);
transactionTemplate.executeWithoutResult(status -> {
int deletedRows = deleteRows(moduleTypes, ids, groupCollectDataIds);
log.info("[module-cleanup] batch: taskIds={}, deletedRows={}, collectedPointers={}, resultObjects={}",
formatIdSample(ids, LOG_SAMPLE_IDS), deletedRows, collected.size(), resultObjectKeys.size());
});
// 远端删除一律放在事务提交之后:事务里做,一旦回滚就会出现「行还在、对象已被删」的悬空引用
submitAndFlush(collected);
deleteResultObjects(resultObjectKeys);
tally.batches++;
tally.deletedTasks += ids.size();
tally.collectedPointers += collected.size();
}
/**
* 收集随行删除的结果文件对象 keyfile_result.result_file_url、task_file_job.result_file_url)。
*
* <p>这些对象此前从不回收——行删掉后再没有任何地方记录过它们,桶里只能靠生命周期规则兜底;
* 一旦某天桶规则被调整(历史上就误配过全桶 30 天过期),就会变成永久垃圾。
*/
private List<String> collectResultObjectKeys(List<String> moduleTypes, List<Long> cleanupTaskIds) {
java.util.Set<String> keys = new java.util.LinkedHashSet<>();
List<FileResultEntity> results = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.in(FileResultEntity::getModuleType, moduleTypes)
.in(FileResultEntity::getTaskId, cleanupTaskIds));
for (FileResultEntity result : results) {
addObjectKey(keys, result.getResultFileUrl());
}
List<TaskFileJobEntity> jobs = taskFileJobMapper.selectList(new LambdaQueryWrapper<TaskFileJobEntity>()
.in(TaskFileJobEntity::getModuleType, moduleTypes)
.in(TaskFileJobEntity::getTaskId, cleanupTaskIds));
for (TaskFileJobEntity job : jobs) {
addObjectKey(keys, job.getResultFileUrl());
}
return new ArrayList<>(keys);
}
private void addObjectKey(java.util.Set<String> keys, String value) {
if (value != null && !value.isBlank()) {
keys.add(value.trim());
}
}
/** 逐个回收结果对象;单个失败只记日志,不影响其余对象与后续批次。 */
private void deleteResultObjects(List<String> objectKeys) {
if (objectKeys.isEmpty()) {
return;
}
int failed = 0;
for (String objectKey : objectKeys) {
try {
ossStorageService.deleteObject(objectKey);
} catch (Exception ex) {
failed++;
log.warn("[module-cleanup] 结果对象删除失败 key={} msg={}", objectKey, ex.getMessage());
}
}
log.info("[module-cleanup] 结果对象回收完成 count={} failed={}", objectKeys.size(), failed);
}
/** 分组拆分后回传计数:批次数、删除任务数、收集指针数。 */
private static final class BatchTally {
private int batches;
private int deletedTasks;
private int collectedPointers;
}
/** 一个事务内完成:删除本批行 → 行删完后提交清理队列并 flush(指针已由调用方收集)。 */
private int deleteRows(List<String> moduleTypes, List<Long> cleanupTaskIds, List<Long> collectDataTaskIds) {
taskFileJobMapper.delete(new LambdaQueryWrapper<TaskFileJobEntity>()
.in(TaskFileJobEntity::getModuleType, moduleTypes)
@@ -257,8 +354,11 @@ public class ModuleHistoryCleanupService {
/**
* 删除前批量收集将随行删除的 payload 指针(chunk.payloadJson、
* scope_state.parsedPayloadJson / stateJson),去重并保持稳定顺序;
* 达到 {@code max} 上限即截断,防止异常巨大的任务行数引发无界收集
* scope_state.parsedPayloadJson / stateJson、result_item.payloadJson、
* result_payload.payloadJson),去重并保持稳定顺序
*
* <p>必须覆盖**所有**存放 transient 指针的列:行一旦删除,漏收的指针就再也无法
* 定位对象,桶里会永久残留孤儿对象。新增写入 payload 指针的列时必须同步加进来。
*/
private List<String> collectPayloadPointers(List<String> moduleTypes, List<Long> cleanupTaskIds, int max) {
java.util.Set<String> pointers = new java.util.LinkedHashSet<>();
@@ -275,8 +375,20 @@ public class ModuleHistoryCleanupService {
collectPointer(pointers, scopeState.getParsedPayloadJson(), max);
collectPointer(pointers, scopeState.getStateJson(), max);
}
List<TaskResultItemEntity> resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.in(TaskResultItemEntity::getModuleType, moduleTypes)
.in(TaskResultItemEntity::getTaskId, cleanupTaskIds));
for (TaskResultItemEntity resultItem : resultItems) {
collectPointer(pointers, resultItem.getPayloadJson(), max);
}
List<TaskResultPayloadEntity> resultPayloads = taskResultPayloadMapper.selectList(new LambdaQueryWrapper<TaskResultPayloadEntity>()
.in(TaskResultPayloadEntity::getModuleType, moduleTypes)
.in(TaskResultPayloadEntity::getTaskId, cleanupTaskIds));
for (TaskResultPayloadEntity resultPayload : resultPayloads) {
collectPointer(pointers, resultPayload.getPayloadJson(), max);
}
if (pointers.size() >= max) {
log.warn("[module-cleanup] payload pointer collection truncated at max={}", max);
log.warn("[module-cleanup] payload pointer collection reached max={}, 交由调用方拆分重试", max);
}
return new ArrayList<>(pointers);
}
@@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.usersecret.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.nanri.aiimage.modules.usersecret.model.entity.UserSecretUsageEntity;
import org.apache.ibatis.annotations.Delete;
import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Mapper;
import org.apache.ibatis.annotations.Param;
@@ -24,4 +25,12 @@ public interface UserSecretUsageMapper extends BaseMapper<UserSecretUsageEntity>
@Param("moduleKey") String moduleKey,
@Param("businessDate") LocalDate businessDate,
@Param("count") int count);
/** 分批删除保留期外的用量日统计行(该表此前只增不删)。 */
@Delete("""
DELETE FROM biz_user_secret_usage_daily
WHERE business_date < #{cutoff}
LIMIT #{batchSize}
""")
int deleteBefore(@Param("cutoff") LocalDate cutoff, @Param("batchSize") int batchSize);
}
@@ -0,0 +1,69 @@
package com.nanri.aiimage.modules.usersecret.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.usersecret.mapper.UserSecretUsageMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.time.LocalDate;
/**
* 密钥用量日统计表(biz_user_secret_usage_daily)的保留期清理。
*
* <p>每次调用按「用户 × 模块 × 业务日」upsert 累加,只增不删;量级是
* 用户数 × 模块数 × 天数,单行很小但长期累积无上限。
*
* <p>保留期取 400 天(而不是常见的一年):用量页要支持同比/跨年对比,
* 恰好一年前的数据仍有价值,留一点余量避免跨年时把刚过期的上年数据删掉。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class UserSecretUsageRetentionService {
/** 单轮最多删除的批次数,剩余留给下一轮。 */
static final int MAX_BATCHES_PER_RUN = 20;
private final UserSecretUsageMapper usageMapper;
private final DistributedJobLockService distributedJobLockService;
@Value("${aiimage.user-secret.usage-retention-days:400}")
private int retentionDays = 400;
@Value("${aiimage.user-secret.usage-retention-batch-size:2000}")
private int retentionBatchSize = 2000;
@Scheduled(cron = "${aiimage.user-secret.usage-retention-cron:0 10 4 * * *}")
public void purgeExpiredUsage() {
int days = Math.max(1, retentionDays);
int batchSize = Math.max(100, retentionBatchSize);
LocalDate cutoff = LocalDate.now().minusDays(days);
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("user-secret:usage-retention", Duration.ofMinutes(15));
if (lockHandle == null) {
log.info("[user-secret] 用量统计保留清理跳过:另一实例持锁");
return;
}
try (lockHandle) {
int totalDeleted = 0;
int batches = 0;
while (batches < MAX_BATCHES_PER_RUN) {
int deleted = usageMapper.deleteBefore(cutoff, batchSize);
batches++;
totalDeleted += deleted;
if (deleted < batchSize) {
break;
}
}
log.info("[user-secret] 用量统计保留清理完成 cutoff={} retentionDays={} deleted={} batches={}",
cutoff, days, totalDeleted, batches);
} catch (Exception ex) {
log.warn("[user-secret] 用量统计保留清理失败 cutoff={} msg={}", cutoff, ex.getMessage(), ex);
}
}
}
@@ -0,0 +1,62 @@
package com.nanri.aiimage.modules.ziniao.memory.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.time.Duration;
/**
* 紫鸟记忆存储(biz_ziniao_memory_store)的过期行清理。
*
* <p>{@link ZiniaoMemoryStoreService#deleteExpired} 早就写好了,但**全库没有任何调用方**——
* 属于典型的"实现了却没接线":过期行只有读取命中时才会被顺手删掉一行,
* 店铺下线、改名后遗留的 key 行永远不会有人再读到,于是永久残留。
*
* <p>这里把它挂上定时任务。单独成类而不是直接给 store 加 {@code @Scheduled}
* store 是被广泛注入的存储组件,不该为了清理任务再依赖分布式锁。
*/
@Service
@RequiredArgsConstructor
@Slf4j
public class ZiniaoMemoryExpiredCleanupService {
/** 单轮最多删除的批次数,剩余留给下一轮。 */
static final int MAX_BATCHES_PER_RUN = 20;
private final ZiniaoMemoryStoreService memoryStoreService;
private final DistributedJobLockService distributedJobLockService;
@Value("${aiimage.ziniao.memory-expired-cleanup-batch-size:500}")
private int batchSize = 500;
@Scheduled(cron = "${aiimage.ziniao.memory-expired-cleanup-cron:0 50 4 * * *}")
public void purgeExpired() {
int size = Math.max(50, batchSize);
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("ziniao:memory-expired-cleanup", Duration.ofMinutes(15));
if (lockHandle == null) {
log.info("[ziniao-memory] 过期行清理跳过:另一实例持锁");
return;
}
try (lockHandle) {
int totalDeleted = 0;
int batches = 0;
while (batches < MAX_BATCHES_PER_RUN) {
int deleted = memoryStoreService.deleteExpired(size);
batches++;
totalDeleted += deleted;
if (deleted < size) {
break;
}
}
log.info("[ziniao-memory] 过期行清理完成 deleted={} batches={}", totalDeleted, batches);
} catch (Exception ex) {
log.warn("[ziniao-memory] 过期行清理失败 msg={}", ex.getMessage(), ex);
}
}
}
@@ -213,6 +213,29 @@ aiimage:
cron: ${AIIMAGE_MODULE_CLEANUP_CRON:0 0 0 * * *}
retention-days: ${AIIMAGE_MODULE_CLEANUP_RETENTION_DAYS:7}
module-types: ${AIIMAGE_MODULE_CLEANUP_MODULE_TYPES:DEDUPE,SPLIT,CONVERT,DELETE_BRAND,PRODUCT_RISK_RESOLVE,PRICE_TRACK,SHOP_MATCH,PATROL_DELETE,QUERY_ASIN,WITHDRAW,APPEARANCE_PATENT,SIMILAR_ASIN,COLLECT_DATA}
# 以下为各业务表的保留期清理(2026-09-16 统一补齐):这些表此前只增不删,
# 或只有"读的时候顺手删一行"这类碰运气的清理。全部按天分批删除、带分布式锁单实例执行。
device-log:
retention-batch-size: ${AIIMAGE_DEVICE_LOG_RETENTION_BATCH_SIZE:2000}
retention-cron: ${AIIMAGE_DEVICE_LOG_RETENTION_CRON:0 40 4 * * *}
# 上架/品牌检测的任务历史:此前完全没有清理,保留期给足业务余量(默认 90 天)。
# 每批只取 50 个任务——上架单任务可达数万行,批量太大会让单次删除事务过长。
publish:
task-retention-days: ${AIIMAGE_PUBLISH_TASK_RETENTION_DAYS:90}
task-retention-batch-size: ${AIIMAGE_PUBLISH_TASK_RETENTION_BATCH_SIZE:50}
task-retention-cron: ${AIIMAGE_PUBLISH_TASK_RETENTION_CRON:0 30 4 * * *}
brand:
task-retention-days: ${AIIMAGE_BRAND_TASK_RETENTION_DAYS:90}
task-retention-batch-size: ${AIIMAGE_BRAND_TASK_RETENTION_BATCH_SIZE:50}
task-retention-cron: ${AIIMAGE_BRAND_TASK_RETENTION_CRON:0 45 4 * * *}
price-track:
loop-run-retention-days: ${AIIMAGE_PRICE_TRACK_LOOP_RUN_RETENTION_DAYS:30}
loop-run-retention-batch-size: ${AIIMAGE_PRICE_TRACK_LOOP_RUN_RETENTION_BATCH_SIZE:500}
loop-run-retention-cron: ${AIIMAGE_PRICE_TRACK_LOOP_RUN_RETENTION_CRON:0 20 4 * * *}
shop-duplicate-check:
scan-retention-days: ${AIIMAGE_SHOP_DUPLICATE_SCAN_RETENTION_DAYS:90}
scan-retention-batch-size: ${AIIMAGE_SHOP_DUPLICATE_SCAN_RETENTION_BATCH_SIZE:200}
scan-retention-cron: ${AIIMAGE_SHOP_DUPLICATE_SCAN_RETENTION_CRON:0 50 3 * * *}
permission-schema-init:
enabled: ${AIIMAGE_PERMISSION_SCHEMA_INIT_ENABLED:false}
task-pressure:
@@ -352,6 +375,10 @@ aiimage:
jikip-user-id: ${AIIMAGE_USER_SECRET_JIKIP_USER_ID:}
# 巡检发现欠费/密钥失效时是否推送站内通知(铃铛)
notify-enabled: ${AIIMAGE_USER_SECRET_NOTIFY_ENABLED:true}
# 用量日统计保留期:取 400 天而非整年,避免跨年时把刚过期的上年数据删掉(含同比对比场景)
usage-retention-days: ${AIIMAGE_USER_SECRET_USAGE_RETENTION_DAYS:400}
usage-retention-batch-size: ${AIIMAGE_USER_SECRET_USAGE_RETENTION_BATCH_SIZE:2000}
usage-retention-cron: ${AIIMAGE_USER_SECRET_USAGE_RETENTION_CRON:0 10 4 * * *}
# 站内通知(铃铛):任务失败扫描 + 下游服务健康探测
notification:
scan-enabled: ${AIIMAGE_NOTIFICATION_SCAN_ENABLED:true}
@@ -375,6 +402,8 @@ aiimage:
maixiang-queue-pending-threshold: ${AIIMAGE_NOTIFICATION_MAIXIANG_QUEUE_PENDING_THRESHOLD:300}
maixiang-queue-processing-threshold: ${AIIMAGE_NOTIFICATION_MAIXIANG_QUEUE_PROCESSING_THRESHOLD:100}
read-retention-days: ${AIIMAGE_NOTIFICATION_READ_RETENTION_DAYS:90}
# 未读通知也设保留期(此前永不清理,不看铃铛的用户会无限累积);给得比已读宽一倍
unread-retention-days: ${AIIMAGE_NOTIFICATION_UNREAD_RETENTION_DAYS:180}
security:
shop-credential-key: ${AIIMAGE_SHOP_CREDENTIAL_KEY:change-me-shop-credential-key}
internal-token: ${AIIMAGE_INTERNAL_TOKEN:}
@@ -406,6 +435,9 @@ aiimage:
open-store-force-download-path: ${AIIMAGE_ZINIAO_OPEN_STORE_FORCE_DOWNLOAD_PATH:}
open-store-extra-args: ${AIIMAGE_ZINIAO_OPEN_STORE_EXTRA_ARGS:--disable-gpu start-maximized}
session-ttl-hours: ${AIIMAGE_ZINIAO_SESSION_TTL_HOURS:2}
# 记忆存储过期行清理:deleteExpired 早已实现但一直无调用方,过期行只有被读到才顺手删一行
memory-expired-cleanup-batch-size: ${AIIMAGE_ZINIAO_MEMORY_EXPIRED_CLEANUP_BATCH_SIZE:500}
memory-expired-cleanup-cron: ${AIIMAGE_ZINIAO_MEMORY_EXPIRED_CLEANUP_CRON:0 50 4 * * *}
shops-cache-minutes: ${AIIMAGE_ZINIAO_SHOPS_CACHE_MINUTES:30}
connect-timeout-seconds: ${AIIMAGE_ZINIAO_CONNECT_TIMEOUT_SECONDS:5}
read-timeout-seconds: ${AIIMAGE_ZINIAO_READ_TIMEOUT_SECONDS:15}
@@ -0,0 +1,117 @@
package com.nanri.aiimage.modules.brand.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.brand.mapper.BrandCrawlTaskMapper;
import org.junit.jupiter.api.Test;
import java.time.LocalDateTime;
import java.util.List;
import java.util.stream.LongStream;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 品牌检测任务保留期清理(brand_crawl_tasks 自建表、不写 biz_file_task,此前没有任何删除路径)。
*/
class BrandTaskRetentionServiceTest {
private final BrandCrawlTaskMapper brandCrawlTaskMapper = mock(BrandCrawlTaskMapper.class);
private final BrandTaskService brandTaskService = mock(BrandTaskService.class);
private final DistributedJobLockService jobLockService = mock(DistributedJobLockService.class);
private BrandTaskRetentionService service() {
return new BrandTaskRetentionService(brandCrawlTaskMapper, brandTaskService, jobLockService);
}
private void lockAvailable() {
when(jobLockService.tryLock(anyString(), any()))
.thenReturn(mock(DistributedJobLockService.LockHandle.class));
}
private List<Long> ids(long from, int count) {
return LongStream.range(from, from + count).boxed().toList();
}
@Test
void deletesInBatchesUntilBatchNotFull() {
lockAvailable();
when(brandCrawlTaskMapper.selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt()))
.thenReturn(ids(1, 50), ids(51, 50), ids(101, 7));
service().purgeExpiredTasks();
verify(brandCrawlTaskMapper, times(3)).selectExpiredTerminalTaskIds(any(LocalDateTime.class), eq(50));
// 逐条走既有删除入口(它已处理任务行 + 存储数据 + 进度缓存)
verify(brandTaskService, times(107)).deleteTask(anyLong());
}
@Test
void stopsAtBatchLimitPerRun() {
lockAvailable();
when(brandCrawlTaskMapper.selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt()))
.thenReturn(ids(1, 50));
service().purgeExpiredTasks();
verify(brandCrawlTaskMapper, times(BrandTaskRetentionService.MAX_BATCHES_PER_RUN))
.selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt());
}
@Test
void skipsWhenAnotherInstanceHoldsLock() {
when(jobLockService.tryLock(anyString(), any())).thenReturn(null);
service().purgeExpiredTasks();
verify(brandCrawlTaskMapper, never()).selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt());
verify(brandTaskService, never()).deleteTask(anyLong());
}
@Test
void singleTaskFailureDoesNotAbortTheBatch() {
// 任务锁被占、任务已在执行中等原因都会让单条删除抛异常,不能因此卡住整轮
lockAvailable();
when(brandCrawlTaskMapper.selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt()))
.thenReturn(List.of(1L, 2L, 3L));
doThrow(new IllegalStateException("锁被占用")).when(brandTaskService).deleteTask(2L);
service().purgeExpiredTasks();
verify(brandTaskService).deleteTask(1L);
verify(brandTaskService).deleteTask(2L);
verify(brandTaskService).deleteTask(3L);
}
@Test
void wholeBatchFailureStopsEarlyToAvoidRepeatingSameBatch() {
lockAvailable();
when(brandCrawlTaskMapper.selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt()))
.thenReturn(ids(1, 50));
doThrow(new IllegalStateException("db down")).when(brandTaskService).deleteTask(anyLong());
service().purgeExpiredTasks();
verify(brandCrawlTaskMapper, times(1)).selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt());
}
@Test
void queryFailureIsLoggedAndDoesNotThrow() {
lockAvailable();
when(brandCrawlTaskMapper.selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt()))
.thenThrow(new IllegalStateException("db down"));
service().purgeExpiredTasks();
verify(brandCrawlTaskMapper).selectExpiredTerminalTaskIds(any(LocalDateTime.class), anyInt());
}
}
@@ -0,0 +1,110 @@
package com.nanri.aiimage.modules.devicelog.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.config.DeviceLogOssProperties;
import com.nanri.aiimage.modules.devicelog.mapper.DeviceLogFileMapper;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import java.time.LocalDate;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 设备日志元数据保留期清理(device_log_file 此前只增不删,对象侧有生命周期而元数据无)。
*/
class DeviceLogRetentionServiceTest {
private final DeviceLogFileMapper fileMapper = mock(DeviceLogFileMapper.class);
private final DistributedJobLockService jobLockService = mock(DistributedJobLockService.class);
private DeviceLogRetentionService service(Integer retentionDays) {
DeviceLogOssProperties properties = new DeviceLogOssProperties();
properties.setRetentionDays(retentionDays);
return new DeviceLogRetentionService(fileMapper, properties, jobLockService);
}
private void lockAvailable() {
when(jobLockService.tryLock(anyString(), any()))
.thenReturn(mock(DistributedJobLockService.LockHandle.class));
}
@Test
void deletesInBatchesUntilBatchNotFull() {
lockAvailable();
when(fileMapper.deleteOlderThanBatch(any(LocalDate.class), anyInt())).thenReturn(2000, 2000, 120);
service(7).purgeExpiredMetadata();
verify(fileMapper, times(3)).deleteOlderThanBatch(any(LocalDate.class), eq(2000));
}
@Test
void stopsAtBatchLimitPerRun() {
lockAvailable();
when(fileMapper.deleteOlderThanBatch(any(LocalDate.class), anyInt())).thenReturn(2000);
service(7).purgeExpiredMetadata();
verify(fileMapper, times(DeviceLogRetentionService.MAX_BATCHES_PER_RUN))
.deleteOlderThanBatch(any(LocalDate.class), anyInt());
}
@Test
void skipsWhenAnotherInstanceHoldsLock() {
when(jobLockService.tryLock(anyString(), any())).thenReturn(null);
service(7).purgeExpiredMetadata();
verify(fileMapper, never()).deleteOlderThanBatch(any(LocalDate.class), anyInt());
}
@Test
void deleteFailureIsLoggedAndDoesNotThrow() {
lockAvailable();
when(fileMapper.deleteOlderThanBatch(any(LocalDate.class), anyInt()))
.thenThrow(new IllegalStateException("db down"));
service(7).purgeExpiredMetadata();
verify(fileMapper).deleteOlderThanBatch(any(LocalDate.class), anyInt());
}
/**
* 契约:cutoff 必须与查询侧可见窗口完全同一条线——page() 展示 log_date >= 今天-(days-1)
* 所以这里只删 log_date < 今天-(days-1),即"早已看不见"的行。
* 天数取自 device-log-oss.retention-days,与对象侧生命周期共用同一配置项。
*/
@Test
void cutoffMatchesVisibleWindowBoundary() {
lockAvailable();
when(fileMapper.deleteOlderThanBatch(any(LocalDate.class), anyInt())).thenReturn(0);
service(7).purgeExpiredMetadata();
ArgumentCaptor<LocalDate> cutoff = ArgumentCaptor.forClass(LocalDate.class);
verify(fileMapper).deleteOlderThanBatch(cutoff.capture(), anyInt());
assertEquals(LocalDate.now().minusDays(6), cutoff.getValue(), "保留 7 天 → 删 7 天前的日期线");
}
@Test
void retentionDaysFallsBackToDefaultWhenUnset() {
lockAvailable();
when(fileMapper.deleteOlderThanBatch(any(LocalDate.class), anyInt())).thenReturn(0);
service(null).purgeExpiredMetadata();
ArgumentCaptor<LocalDate> cutoff = ArgumentCaptor.forClass(LocalDate.class);
verify(fileMapper).deleteOlderThanBatch(cutoff.capture(), anyInt());
assertEquals(LocalDate.now().minusDays(6), cutoff.getValue(), "未配置时回退默认 7 天");
}
}
@@ -0,0 +1,76 @@
package com.nanri.aiimage.modules.pricetrack.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.pricetrack.mapper.PriceTrackLoopRunMapper;
import org.junit.jupiter.api.Test;
import java.time.LocalDateTime;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 跟价循环批次表保留期清理(该表此前有 insert 无 delete,属纯遗漏)。
*/
class PriceTrackLoopRunRetentionServiceTest {
private final PriceTrackLoopRunMapper loopRunMapper = mock(PriceTrackLoopRunMapper.class);
private final DistributedJobLockService jobLockService = mock(DistributedJobLockService.class);
private PriceTrackLoopRunRetentionService service() {
return new PriceTrackLoopRunRetentionService(loopRunMapper, jobLockService);
}
private void lockAvailable() {
when(jobLockService.tryLock(anyString(), any()))
.thenReturn(mock(DistributedJobLockService.LockHandle.class));
}
@Test
void deletesInBatchesUntilBatchNotFull() {
lockAvailable();
when(loopRunMapper.deleteFinishedBefore(any(LocalDateTime.class), anyInt())).thenReturn(500, 500, 30);
service().purgeExpiredLoopRuns();
verify(loopRunMapper, times(3)).deleteFinishedBefore(any(LocalDateTime.class), eq(500));
}
@Test
void stopsAtBatchLimitPerRun() {
lockAvailable();
when(loopRunMapper.deleteFinishedBefore(any(LocalDateTime.class), anyInt())).thenReturn(500);
service().purgeExpiredLoopRuns();
verify(loopRunMapper, times(PriceTrackLoopRunRetentionService.MAX_BATCHES_PER_RUN))
.deleteFinishedBefore(any(LocalDateTime.class), anyInt());
}
@Test
void skipsWhenAnotherInstanceHoldsLock() {
when(jobLockService.tryLock(anyString(), any())).thenReturn(null);
service().purgeExpiredLoopRuns();
verify(loopRunMapper, never()).deleteFinishedBefore(any(LocalDateTime.class), anyInt());
}
@Test
void deleteFailureIsLoggedAndDoesNotThrow() {
lockAvailable();
when(loopRunMapper.deleteFinishedBefore(any(LocalDateTime.class), anyInt()))
.thenThrow(new IllegalStateException("db down"));
service().purgeExpiredLoopRuns();
verify(loopRunMapper).deleteFinishedBefore(any(LocalDateTime.class), anyInt());
}
}
@@ -0,0 +1,116 @@
package com.nanri.aiimage.modules.publish.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
import org.junit.jupiter.api.Test;
import java.time.LocalDateTime;
import java.util.List;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 上架任务保留期清理(PUBLISH 此前完全没有清理,biz_publish_item 单任务可达数万行)。
*/
class PublishTaskRetentionServiceTest {
private final FileTaskMapper fileTaskMapper = mock(FileTaskMapper.class);
private final PublishTaskService publishTaskService = mock(PublishTaskService.class);
private final DistributedJobLockService jobLockService = mock(DistributedJobLockService.class);
private PublishTaskRetentionService service() {
return new PublishTaskRetentionService(fileTaskMapper, publishTaskService, jobLockService);
}
private void lockAvailable() {
when(jobLockService.tryLock(anyString(), any()))
.thenReturn(mock(DistributedJobLockService.LockHandle.class));
}
private List<Long> ids(long from, int count) {
return java.util.stream.LongStream.range(from, from + count).boxed().toList();
}
@Test
void deletesInBatchesUntilBatchNotFull() {
lockAvailable();
when(fileTaskMapper.selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), anyInt()))
.thenReturn(ids(1, 50), ids(51, 50), ids(101, 7));
service().purgeExpiredTasks();
verify(fileTaskMapper, times(3)).selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), eq(50));
verify(publishTaskService, times(107)).deleteTaskForRetention(anyLong());
}
@Test
void stopsAtBatchLimitPerRun() {
lockAvailable();
when(fileTaskMapper.selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), anyInt()))
.thenReturn(ids(1, 50));
service().purgeExpiredTasks();
verify(fileTaskMapper, times(PublishTaskRetentionService.MAX_BATCHES_PER_RUN))
.selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), anyInt());
}
@Test
void skipsWhenAnotherInstanceHoldsLock() {
when(jobLockService.tryLock(anyString(), any())).thenReturn(null);
service().purgeExpiredTasks();
verify(fileTaskMapper, never()).selectExpiredTerminalTaskIds(anyString(), any(LocalDateTime.class), anyInt());
verify(publishTaskService, never()).deleteTaskForRetention(anyLong());
}
@Test
void singleTaskFailureDoesNotAbortTheBatch() {
// 一个坏任务不能卡住整轮:失败的只计数,同批其余任务照常删除
lockAvailable();
when(fileTaskMapper.selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), anyInt()))
.thenReturn(List.of(1L, 2L, 3L));
doThrow(new IllegalStateException("delete failed")).when(publishTaskService).deleteTaskForRetention(2L);
service().purgeExpiredTasks();
verify(publishTaskService).deleteTaskForRetention(1L);
verify(publishTaskService).deleteTaskForRetention(2L);
verify(publishTaskService).deleteTaskForRetention(3L);
}
@Test
void wholeBatchFailureStopsEarlyToAvoidRepeatingSameBatch() {
// 整批全失败说明是环境性故障(不是单个坏任务),继续循环只会反复重试同一批
lockAvailable();
when(fileTaskMapper.selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), anyInt()))
.thenReturn(ids(1, 50));
doThrow(new IllegalStateException("db down")).when(publishTaskService).deleteTaskForRetention(anyLong());
service().purgeExpiredTasks();
verify(fileTaskMapper, times(1)).selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), anyInt());
}
@Test
void queryFailureIsLoggedAndDoesNotThrow() {
lockAvailable();
when(fileTaskMapper.selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), anyInt()))
.thenThrow(new IllegalStateException("db down"));
service().purgeExpiredTasks();
verify(fileTaskMapper).selectExpiredTerminalTaskIds(eq("PUBLISH"), any(LocalDateTime.class), anyInt());
}
}
@@ -0,0 +1,97 @@
package com.nanri.aiimage.modules.shopduplicatecheck.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.shopduplicatecheck.mapper.ShopDataDuplicateScanMapper;
import org.junit.jupiter.api.Test;
import java.time.LocalDateTime;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import org.mockito.ArgumentCaptor;
/**
* 撞款扫描结果表保留期清理(读取侧只认最新一行,历史行此前无删除路径)。
*/
class ShopDataDuplicateScanRetentionServiceTest {
private final ShopDataDuplicateScanMapper scanMapper = mock(ShopDataDuplicateScanMapper.class);
private final DistributedJobLockService jobLockService = mock(DistributedJobLockService.class);
private ShopDataDuplicateScanRetentionService service() {
return new ShopDataDuplicateScanRetentionService(scanMapper, jobLockService);
}
private void lockAvailable() {
when(jobLockService.tryLock(anyString(), any()))
.thenReturn(mock(DistributedJobLockService.LockHandle.class));
}
@Test
void deletesInBatchesUntilBatchNotFull() {
lockAvailable();
when(scanMapper.selectNthNewestId(anyInt())).thenReturn(900L);
when(scanMapper.deleteOlderThanBatch(any(LocalDateTime.class), anyLong(), anyInt()))
.thenReturn(200, 200, 40);
service().purgeExpiredScans();
verify(scanMapper, times(3)).deleteOlderThanBatch(any(LocalDateTime.class), anyLong(), eq(200));
}
@Test
void stopsAtBatchLimitPerRun() {
lockAvailable();
when(scanMapper.selectNthNewestId(anyInt())).thenReturn(900L);
when(scanMapper.deleteOlderThanBatch(any(LocalDateTime.class), anyLong(), anyInt())).thenReturn(200);
service().purgeExpiredScans();
verify(scanMapper, times(ShopDataDuplicateScanRetentionService.MAX_BATCHES_PER_RUN))
.deleteOlderThanBatch(any(LocalDateTime.class), anyLong(), anyInt());
}
@Test
void skipsWhenAnotherInstanceHoldsLock() {
when(jobLockService.tryLock(anyString(), any())).thenReturn(null);
service().purgeExpiredScans();
verify(scanMapper, never()).deleteOlderThanBatch(any(LocalDateTime.class), anyLong(), anyInt());
}
@Test
void deleteFailureIsLoggedAndDoesNotThrow() {
lockAvailable();
when(scanMapper.selectNthNewestId(anyInt())).thenReturn(900L);
when(scanMapper.deleteOlderThanBatch(any(LocalDateTime.class), anyLong(), anyInt()))
.thenThrow(new IllegalStateException("db down"));
service().purgeExpiredScans();
verify(scanMapper).deleteOlderThanBatch(any(LocalDateTime.class), anyLong(), anyInt());
}
/** 行数不足保护量时不设保护线(Long.MAX_VALUE),此时只按时间线删,不会误伤最新几行。 */
@Test
void keepsAllRowsWhenFewerThanProtectedCount() {
lockAvailable();
when(scanMapper.selectNthNewestId(anyInt())).thenReturn(null);
when(scanMapper.deleteOlderThanBatch(any(LocalDateTime.class), anyLong(), anyInt())).thenReturn(0);
service().purgeExpiredScans();
ArgumentCaptor<Long> protectLine = ArgumentCaptor.forClass(Long.class);
verify(scanMapper).deleteOlderThanBatch(any(LocalDateTime.class), protectLine.capture(), anyInt());
assertTrue(protectLine.getValue() == Long.MAX_VALUE, "保护线为 Long.MAX_VALUE,等于不额外设限");
}
}
@@ -167,7 +167,7 @@ class ModuleHistoryCleanupKeysetPaginationTest {
cleanupProperties, fileTaskMapper, fileResultMapper, taskFileJobMapper,
taskResultItemMapper, taskProgressSnapshotMapper, taskResultPayloadMapper,
taskScopeStateMapper, taskChunkMapper, collectDataItemMapper, lockService, orchestrator,
transactionManager);
oss, transactionManager);
}
private static FileTaskEntity expiredTask(long id) {
@@ -158,7 +158,7 @@ class ModuleHistoryCleanupLoggingTest {
cleanupProperties, fileTaskMapper, fileResultMapper, taskFileJobMapper,
taskResultItemMapper, taskProgressSnapshotMapper, taskResultPayloadMapper,
taskScopeStateMapper, taskChunkMapper, collectDataItemMapper, lockService, orchestrator,
transactionManager);
oss, transactionManager);
}
private static FileTaskEntity expiredTask(long id) {
@@ -56,6 +56,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -148,7 +149,7 @@ class ModuleHistoryCleanupPayloadCleanupTest {
cleanupProperties, fileTaskMapper, fileResultMapper, taskFileJobMapper,
taskResultItemMapper, taskProgressSnapshotMapper, taskResultPayloadMapper,
taskScopeStateMapper, taskChunkMapper, collectDataItemMapper, lockService, orchestrator,
transactionManager);
oss, transactionManager);
}
private static FileTaskEntity expiredTask(long id) {
@@ -225,6 +226,38 @@ class ModuleHistoryCleanupPayloadCleanupTest {
return state;
}
private static TaskResultItemEntity resultItem(long taskId, String payloadJson) {
TaskResultItemEntity item = new TaskResultItemEntity();
item.setTaskId(taskId);
item.setModuleType("DEDUPE");
item.setPayloadJson(payloadJson);
return item;
}
private static TaskResultPayloadEntity resultPayload(long taskId, String payloadJson) {
TaskResultPayloadEntity payload = new TaskResultPayloadEntity();
payload.setTaskId(taskId);
payload.setModuleType("DEDUPE");
payload.setPayloadJson(payloadJson);
return payload;
}
private static FileResultEntity resultRow(long taskId, String resultFileUrl) {
FileResultEntity row = new FileResultEntity();
row.setTaskId(taskId);
row.setModuleType("DEDUPE");
row.setResultFileUrl(resultFileUrl);
return row;
}
private static TaskFileJobEntity resultJob(long taskId, String resultFileUrl) {
TaskFileJobEntity job = new TaskFileJobEntity();
job.setTaskId(taskId);
job.setModuleType("DEDUPE");
job.setResultFileUrl(resultFileUrl);
return job;
}
private void stubDeletes() {
when(taskFileJobMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
when(taskResultItemMapper.delete(any(LambdaQueryWrapper.class))).thenReturn(0);
@@ -355,7 +388,7 @@ class ModuleHistoryCleanupPayloadCleanupTest {
@Test
void test_task_069_payload_cleanup_boundary_limit_and_overflow() throws Exception {
// 上限/超限:收集达到上限即停止,不发生无界收集;
// flush 时引用检查失败则本批保留(保守),可下次重试
// 单任务即触顶、无法再拆分时整组不删——宁可留脏行待下轮重试,也不让指针随行消失造成孤儿对象
stubExpiredTasks(51);
stubDeletes();
ReflectionTestUtils.setField(cleanupService, "maxCollectPayloadsPerRun", 3);
@@ -363,14 +396,36 @@ class ModuleHistoryCleanupPayloadCleanupTest {
for (int i = 1; i <= 5; i++) {
chunks.add(chunk(51, jsonPointer(50 + i)));
}
doAnswer(sequenceThenFail(chunks, 1)).when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class));
doAnswer(sequence(chunks, 1, List.of())).when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class));
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
cleanupService.cleanupConfiguredModules();
assertEquals(3, orchestrator.pendingCount(), "仅保留上限内的指针");
assertEquals(0, orchestrator.pendingCount(), "触顶且无法拆分时不提交任何指针");
verify(rustfs, never()).deleteObject(anyString());
verify(taskChunkMapper).delete(any(LambdaQueryWrapper.class));
verify(taskChunkMapper, never()).delete(any(LambdaQueryWrapper.class));
}
@Test
void test_payload_cleanup_overflow_splits_group_instead_of_truncating() throws Exception {
// 触顶拆分:整组 4 个指针、上限 3 时不得"截断后照样把整组删掉"——
// 那会让未收集到的指针随行一起消失,对象再无引用可查、永久孤儿化;
// 正确行为是二分后分两批处理。两次 delete 即证明整组未被当作一批删除。
stubExpiredTasks(81, 82);
stubDeletes();
ReflectionTestUtils.setField(cleanupService, "maxCollectPayloadsPerRun", 3);
List<TaskChunkEntity> all = List.of(
chunk(81, pointer(81)), chunk(81, pointer(82)),
chunk(82, pointer(83)), chunk(82, pointer(84)));
doAnswer(sequence(all, 1, List.of())).when(taskChunkMapper).selectList(any(LambdaQueryWrapper.class));
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
CountDownLatch done = latchOnDelete(1);
cleanupService.cleanupConfiguredModules();
verify(taskChunkMapper, times(2)).delete(any(LambdaQueryWrapper.class));
assertEquals(1, done.getCount(), "拆分后的两批均未携带被截断的指针,无对象进入删除队列");
}
@Test
@@ -392,6 +447,69 @@ class ModuleHistoryCleanupPayloadCleanupTest {
verify(rustfs, never()).deleteObject(anyString());
}
@Test
void test_payload_cleanup_collects_result_item_and_result_payload_pointers() throws Exception {
// 回归:result_item / result_payload 的 payloadJson 同样存放 transient 指针
//storeResultItemPayload / storeResultPayload),行删除前必须一并收集,
// 否则这两类对象随行消失、永久孤儿化在桶里。
stubExpiredTasks(91);
stubDeletes();
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
doAnswer(sequence(List.of(resultItem(91, pointer(91))), 1, List.of()))
.when(taskResultItemMapper).selectList(any(LambdaQueryWrapper.class));
doAnswer(sequence(List.of(resultPayload(91, jsonPointer(92))), 1, List.of()))
.when(taskResultPayloadMapper).selectList(any(LambdaQueryWrapper.class));
CountDownLatch done = latchOnDelete(2);
cleanupService.cleanupConfiguredModules();
assertTrue(done.await(2, TimeUnit.SECONDS), "result_item / result_payload 指针各自异步删除一次");
verify(rustfs, times(2)).deleteObject(anyString());
verify(rustfs).deleteObject("task-parsed/test/91/scope/latest.json");
verify(rustfs).deleteObject("task-parsed/test/92/scope/latest.json");
}
@Test
void test_payload_cleanup_recycles_result_file_objects() throws Exception {
// 结果对象回收:file_result / task_file_job 的 result_file_url 指向的对象随行删除一并回收。
// 此前只删行、对象永久留在桶里(只能靠桶生命周期兜底)。
// 两个来源指向同一个 key 时只删一次(去重)。
stubExpiredTasks(101);
stubDeletes();
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
when(fileResultMapper.selectList(any(LambdaQueryWrapper.class)))
.thenReturn(List.of(resultRow(101, "result/dedupe/101.xlsx")));
when(taskFileJobMapper.selectList(any(LambdaQueryWrapper.class)))
.thenReturn(List.of(resultJob(101, "result/dedupe/101.xlsx")));
cleanupService.cleanupConfiguredModules();
verify(oss).deleteObject("result/dedupe/101.xlsx");
verify(oss, times(1)).deleteObject(anyString());
verify(fileTaskMapper).delete(any(LambdaQueryWrapper.class));
}
@Test
void test_payload_cleanup_result_object_failure_does_not_abort_batch() throws Exception {
// 单个结果对象删不掉只记日志:后续对象照常回收,行照常删除,不影响下一批。
stubExpiredTasks(102);
stubDeletes();
when(taskChunkMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
when(taskScopeStateMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
when(fileResultMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of(
resultRow(102, "result/dedupe/102a.xlsx"),
resultRow(102, "result/dedupe/102b.xlsx")));
when(taskFileJobMapper.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of());
doThrow(new IllegalStateException("oss down")).when(oss).deleteObject("result/dedupe/102a.xlsx");
cleanupService.cleanupConfiguredModules();
verify(oss).deleteObject("result/dedupe/102b.xlsx");
verify(fileTaskMapper).delete(any(LambdaQueryWrapper.class));
}
@Test
void test_task_069_payload_cleanup_dependency_failure_releases_resources() throws Exception {
// 依赖失败:行删除抛异常时中止,指针不提交、不 flush、不删除对象;
@@ -0,0 +1,76 @@
package com.nanri.aiimage.modules.usersecret.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import com.nanri.aiimage.modules.usersecret.mapper.UserSecretUsageMapper;
import org.junit.jupiter.api.Test;
import java.time.LocalDate;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 密钥用量日统计表保留期清理(该表按「用户 × 模块 × 日」只增不删)。
*/
class UserSecretUsageRetentionServiceTest {
private final UserSecretUsageMapper usageMapper = mock(UserSecretUsageMapper.class);
private final DistributedJobLockService jobLockService = mock(DistributedJobLockService.class);
private UserSecretUsageRetentionService service() {
return new UserSecretUsageRetentionService(usageMapper, jobLockService);
}
private void lockAvailable() {
when(jobLockService.tryLock(anyString(), any()))
.thenReturn(mock(DistributedJobLockService.LockHandle.class));
}
@Test
void deletesInBatchesUntilBatchNotFull() {
lockAvailable();
when(usageMapper.deleteBefore(any(LocalDate.class), anyInt())).thenReturn(2000, 2000, 300);
service().purgeExpiredUsage();
verify(usageMapper, times(3)).deleteBefore(any(LocalDate.class), eq(2000));
}
@Test
void stopsAtBatchLimitPerRun() {
lockAvailable();
when(usageMapper.deleteBefore(any(LocalDate.class), anyInt())).thenReturn(2000);
service().purgeExpiredUsage();
verify(usageMapper, times(UserSecretUsageRetentionService.MAX_BATCHES_PER_RUN))
.deleteBefore(any(LocalDate.class), anyInt());
}
@Test
void skipsWhenAnotherInstanceHoldsLock() {
when(jobLockService.tryLock(anyString(), any())).thenReturn(null);
service().purgeExpiredUsage();
verify(usageMapper, never()).deleteBefore(any(LocalDate.class), anyInt());
}
@Test
void deleteFailureIsLoggedAndDoesNotThrow() {
lockAvailable();
when(usageMapper.deleteBefore(any(LocalDate.class), anyInt()))
.thenThrow(new IllegalStateException("db down"));
service().purgeExpiredUsage();
verify(usageMapper).deleteBefore(any(LocalDate.class), anyInt());
}
}
@@ -0,0 +1,72 @@
package com.nanri.aiimage.modules.ziniao.memory.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import org.junit.jupiter.api.Test;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 紫鸟记忆存储过期行清理(deleteExpired 此前无任何调用方,属"实现了没接线")。
*/
class ZiniaoMemoryExpiredCleanupServiceTest {
private final ZiniaoMemoryStoreService memoryStoreService = mock(ZiniaoMemoryStoreService.class);
private final DistributedJobLockService jobLockService = mock(DistributedJobLockService.class);
private ZiniaoMemoryExpiredCleanupService service() {
return new ZiniaoMemoryExpiredCleanupService(memoryStoreService, jobLockService);
}
private void lockAvailable() {
when(jobLockService.tryLock(anyString(), any()))
.thenReturn(mock(DistributedJobLockService.LockHandle.class));
}
@Test
void deletesInBatchesUntilBatchNotFull() {
lockAvailable();
when(memoryStoreService.deleteExpired(anyInt())).thenReturn(500, 500, 20);
service().purgeExpired();
verify(memoryStoreService, times(3)).deleteExpired(eq(500));
}
@Test
void stopsAtBatchLimitPerRun() {
lockAvailable();
when(memoryStoreService.deleteExpired(anyInt())).thenReturn(500);
service().purgeExpired();
verify(memoryStoreService, times(ZiniaoMemoryExpiredCleanupService.MAX_BATCHES_PER_RUN))
.deleteExpired(anyInt());
}
@Test
void skipsWhenAnotherInstanceHoldsLock() {
when(jobLockService.tryLock(anyString(), any())).thenReturn(null);
service().purgeExpired();
verify(memoryStoreService, never()).deleteExpired(anyInt());
}
@Test
void deleteFailureIsLoggedAndDoesNotThrow() {
lockAvailable();
when(memoryStoreService.deleteExpired(anyInt())).thenThrow(new IllegalStateException("db down"));
service().purgeExpired();
verify(memoryStoreService).deleteExpired(anyInt());
}
}