From f2ada023830809bc5339ba17916f1d295aba0699 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Wed, 16 Sep 2026 01:59:28 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E4=BF=9D=E7=95=99=E6=9C=9F=E6=B8=85?= =?UTF-8?q?=E7=90=86):=20=E8=A1=A5=E9=BD=90=207=20=E5=A4=84=E5=8F=AA?= =?UTF-8?q?=E5=A2=9E=E4=B8=8D=E5=88=A0=E7=9A=84=E6=95=B0=E6=8D=AE=EF=BC=9B?= =?UTF-8?q?=E4=BF=AE=E7=BB=93=E6=9E=9C=E5=AF=B9=E8=B1=A1=E5=AD=A4=E5=84=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 审核发现一批表/元数据只增不删，且对象存储存在活跃泄漏： - 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 + 分布式锁单实例 + 分批 + 单轮批次数上限 + 失败只记日志。 --- .../config/NotificationProperties.java | 8 + .../brand/mapper/BrandCrawlTaskMapper.java | 24 +++ .../service/BrandTaskRetentionService.java | 93 +++++++++++ .../devicelog/mapper/DeviceLogFileMapper.java | 17 ++ .../service/DeviceLogRetentionService.java | 77 +++++++++ .../service/NotificationScanScheduler.java | 8 +- .../service/NotificationService.java | 11 ++ .../mapper/PriceTrackLoopRunMapper.java | 20 ++- .../PriceTrackLoopRunRetentionService.java | 70 ++++++++ .../service/PublishTaskRetentionService.java | 97 +++++++++++ .../publish/service/PublishTaskService.java | 42 +++++ .../mapper/ShopDataDuplicateScanMapper.java | 19 +++ ...ShopDataDuplicateScanRetentionService.java | 76 +++++++++ .../modules/task/mapper/FileTaskMapper.java | 28 ++++ .../service/ModuleHistoryCleanupService.java | 154 +++++++++++++++--- .../mapper/UserSecretUsageMapper.java | 9 + .../UserSecretUsageRetentionService.java | 69 ++++++++ .../ZiniaoMemoryExpiredCleanupService.java | 62 +++++++ .../src/main/resources/application.yml | 32 ++++ .../BrandTaskRetentionServiceTest.java | 117 +++++++++++++ .../DeviceLogRetentionServiceTest.java | 110 +++++++++++++ ...PriceTrackLoopRunRetentionServiceTest.java | 76 +++++++++ .../PublishTaskRetentionServiceTest.java | 116 +++++++++++++ ...DataDuplicateScanRetentionServiceTest.java | 97 +++++++++++ ...uleHistoryCleanupKeysetPaginationTest.java | 2 +- .../ModuleHistoryCleanupLoggingTest.java | 2 +- ...oduleHistoryCleanupPayloadCleanupTest.java | 128 ++++++++++++++- .../UserSecretUsageRetentionServiceTest.java | 76 +++++++++ ...ZiniaoMemoryExpiredCleanupServiceTest.java | 72 ++++++++ 29 files changed, 1680 insertions(+), 32 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskRetentionService.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/devicelog/service/DeviceLogRetentionService.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunRetentionService.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskRetentionService.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/shopduplicatecheck/service/ShopDataDuplicateScanRetentionService.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/usersecret/service/UserSecretUsageRetentionService.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryExpiredCleanupService.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/brand/service/BrandTaskRetentionServiceTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/devicelog/service/DeviceLogRetentionServiceTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunRetentionServiceTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishTaskRetentionServiceTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/shopduplicatecheck/service/ShopDataDuplicateScanRetentionServiceTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/usersecret/service/UserSecretUsageRetentionServiceTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryExpiredCleanupServiceTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/NotificationProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/NotificationProperties.java index fb032b3d..61a10f79 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/NotificationProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/NotificationProperties.java @@ -67,4 +67,12 @@ public class NotificationProperties { /** 已读通知保留天数（超期自动清理），默认 90 天。 */ private int readRetentionDays = 90; + + /** + * 未读通知保留天数，默认 180 天（比已读长一倍）。 + * + *

未读通知此前永不清理：不看铃铛的用户会无限累积。保留期给得更宽， + * 是因为未读意味着"用户可能还没看到"，但也不能永远留着。 + */ + private int unreadRetentionDays = 180; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/mapper/BrandCrawlTaskMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/mapper/BrandCrawlTaskMapper.java index 50afef94..a4c7f964 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/mapper/BrandCrawlTaskMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/mapper/BrandCrawlTaskMapper.java @@ -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 { + + /** + * 查询超过保留期的终态品牌检测任务 id（保留期清理用，只取 id 不拉整行——历史行的 + * file_paths/result_paths JSON 字段可能很大）。 + * + *

终态集合与 {@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 selectExpiredTerminalTaskIds(@Param("cutoff") LocalDateTime cutoff, + @Param("batchSize") int batchSize); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskRetentionService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskRetentionService.java new file mode 100644 index 00000000..097b144d --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskRetentionService.java @@ -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 审核：该表只增不删，永久累积）。 + * + *

该表自建、不写 biz_file_task，故不在 ModuleHistoryCleanupService 的清理名单里， + * 此前没有任何删除路径。这里只查过期终态任务的 id，逐个走 + * {@link BrandTaskService#deleteTask(Long)} 既有删除入口 —— 它已处理任务行删除 + + * 存储数据清理（brandTaskStorageService.deleteTaskData）+ 进度缓存清理，本类不重新实现删除逻辑。 + * + *

双节点用 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 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); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/devicelog/mapper/DeviceLogFileMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/devicelog/mapper/DeviceLogFileMapper.java index 3bc3f583..28213410 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/devicelog/mapper/DeviceLogFileMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/devicelog/mapper/DeviceLogFileMapper.java @@ -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 { + + /** + * 按日志日期分批删除保留期外的元数据行。 + * + *

只删 {@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); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/devicelog/service/DeviceLogRetentionService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/devicelog/service/DeviceLogRetentionService.java new file mode 100644 index 00000000..3ec06f81 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/devicelog/service/DeviceLogRetentionService.java @@ -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; + +/** + * 设备日志元数据的保留期清理。 + * + *

对象侧一直有生命周期（主机B 独立 MinIO 桶，由运维 mc 定时任务按 7 天回收）， + * 但 {@code device_log_file} 的元数据行此前**只增不删**：客户端每 60 秒上报一轮， + * 每台设备每天登记若干「来源+设备+文件名+日期」行，长期运行会无限累积。 + * + *

查询侧本就只展示保留期内的行（{@link DeviceLogService#page} 用同一个 + * {@code retentionDays} 过滤），这里按完全相同的边界删掉早已不可见的行。 + * 保留天数直接复用 {@code aiimage.device-log-oss.retention-days}， + * 与对象侧共用同一个配置项，避免两边各写一份天数后悄悄漂移。 + * + *

对象本身仍由桶生命周期负责回收，本任务不碰对象：两条时间线按「对象修改时间」 + * 与「日志日期」衡量，可能有几天错位，但对象最终仍会被桶规则删除，不会永久残留。 + * + *

双节点用 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); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationScanScheduler.java b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationScanScheduler.java index bfe7c67c..1aaf387c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationScanScheduler.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationScanScheduler.java @@ -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) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationService.java index 353c7f44..c22a8567 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/notification/service/NotificationService.java @@ -217,6 +217,17 @@ public class NotificationService { return deleted; } + /** 清理指定时间之前仍未读的通知（保留期由调用方决定，比已读给得更宽）。 */ + public int cleanupUnreadBefore(LocalDateTime cutoff) { + int deleted = userNotificationMapper.delete(new LambdaQueryWrapper() + .isNull(UserNotificationEntity::getReadAt) + .lt(UserNotificationEntity::getCreatedAt, cutoff)); + if (deleted > 0) { + log.info("[notification] 清理历史未读通知 cutoff={} 删除={} 条", cutoff, deleted); + } + return deleted; + } + private LambdaQueryWrapper baseWrapper(Long userId, String audience, boolean onlyUnread, NotificationPageQuery query) { LambdaQueryWrapper wrapper = new LambdaQueryWrapper() diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/mapper/PriceTrackLoopRunMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/mapper/PriceTrackLoopRunMapper.java index 05760a31..09b9a10c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/mapper/PriceTrackLoopRunMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/mapper/PriceTrackLoopRunMapper.java @@ -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 { -} + + /** + * 分批删除已结束且超过保留期的循环批次行。 + * + *

只删终态行：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); +} \ No newline at end of file diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunRetentionService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunRetentionService.java new file mode 100644 index 00000000..1c4fbfde --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunRetentionService.java @@ -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）的保留期清理。 + * + *

每次「循环跟价」插一行，且只在请求停止时更新——全模块此前没有任何删除路径（纯遗漏）。 + * 读取侧只有两个入口：按 id 取单条（用户当下正在看的那一轮）与「当前是否有活跃循环」， + * 没有任何历史列表查询，所以终态批次过了保留期即可安全删除。 + * + *

只删终态（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); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskRetentionService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskRetentionService.java new file mode 100644 index 00000000..889fc82b --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskRetentionService.java @@ -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 的任务历史此前完全没有清理，永久累积）。 + * + *

涉及 biz_file_task / biz_file_result / biz_publish_item（每个上传 Excel 的每一行落一行， + * 单任务可达数万行）/ biz_publish_file，此前只有用户手动删任务才回收。这里刻意不复用 + * ModuleHistoryCleanupService：上架单任务行数太大，且删除必须经过业务删除入口回收结果对象。 + * + *

删除动作复用 {@link PublishTaskService#deleteTaskForRetention(Long)}（与用户删任务 + * 同一套删除实现：明细/结果/分片载荷 + 事务提交后回收结果对象），本类只做「查一批过期 id → + * 逐个调用 → 计数」。该入口保留状态校验，只删终态任务，PENDING/RUNNING 不会被碰。 + * + *

双节点用 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 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); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java index e1e916eb..756960f9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/publish/service/PublishTaskService.java @@ -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 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); + } + + /** + * 保留期清理专用删除入口（内部调用，定时任务没有用户身份）。 + * + *

与 {@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 results = fileResultMapper.selectList(new LambdaQueryWrapper() .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 normalizeTaskIds(List taskIds) { if (taskIds == null) { return List.of(); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopduplicatecheck/mapper/ShopDataDuplicateScanMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopduplicatecheck/mapper/ShopDataDuplicateScanMapper.java index 5757bbff..d8242ed3 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopduplicatecheck/mapper/ShopDataDuplicateScanMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopduplicatecheck/mapper/ShopDataDuplicateScanMapper.java @@ -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 { @@ -20,4 +24,19 @@ public interface ShopDataDuplicateScanMapper extends BaseMapper扫描每晚自动跑一轮、管理员还能手动触发，每次插入一行**含整份聚合 payload 的结果**； + * 但读取侧只认最新一行（{@code selectLatestLightRow} / {@code selectLatestFullRow} 都是 + * {@code ORDER BY id DESC LIMIT 1}），历史行纯属占用磁盘，此前无任何删除路径。 + * + *

保护策略：无论多旧，始终保留按 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); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/FileTaskMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/FileTaskMapper.java index 7c3c1334..83dac680 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/FileTaskMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/FileTaskMapper.java @@ -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 { + + /** + * 查询某模块下超过保留期的终态任务 id（保留期清理用，只取 id 不拉整行）。 + * + *

终态集合与 ModuleHistoryCleanupService 的 TERMINAL_STATUSES 一致：PENDING/RUNNING 绝不返回—— + * 正在跑的任务被删会让分片回传、结果组装找不到任务行。 + * + *

时间线只用 updated_at：该列由 DB 的 ON UPDATE CURRENT_TIMESTAMP 维护，任何一次行更新 + * （含终态写入）都会刷新，不会早于 finished_at（finished_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 selectExpiredTerminalTaskIds(@Param("moduleType") String moduleType, + @Param("cutoff") LocalDateTime cutoff, + @Param("batchSize") int batchSize); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java index a20ecd78..ba5ae6f3 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupService.java @@ -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 taskIds = batchTaskIds; - final List collectDataTaskIds = batchCollectDataTaskIds; - final List types = moduleTypes; - final List 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 指针 → 删除行 → 提交对象回收。 + * + *

指针数触顶时把任务组二分拆分重试，保证「凡被删除的行，其 transient 指针一定已进回收队列」—— + * 截断会让指针随行一起消失，对象再无引用可查、永久留在桶里。 + * 拆到单个任务仍触顶则整组不删并记 error（宁可留脏行待下轮重试，也不制造孤儿对象）； + * 运维可据此调大 {@code aiimage.module-cleanup.max-collect-payloads} 后自动收敛。 + */ + private void cleanupTaskGroup(List moduleTypes, List taskIds, List collectDataTaskIds, + int depth, BatchTally tally) { + if (taskIds.isEmpty()) { + return; + } + List 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 ids = List.copyOf(taskIds); + final Set collectDataIdSet = Set.copyOf(collectDataTaskIds); + final List groupCollectDataIds = ids.stream().filter(collectDataIdSet::contains).toList(); + List 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(); + } + + /** + * 收集随行删除的结果文件对象 key（file_result.result_file_url、task_file_job.result_file_url）。 + * + *

这些对象此前从不回收——行删掉后再没有任何地方记录过它们，桶里只能靠生命周期规则兜底； + * 一旦某天桶规则被调整（历史上就误配过全桶 30 天过期），就会变成永久垃圾。 + */ + private List collectResultObjectKeys(List moduleTypes, List cleanupTaskIds) { + java.util.Set keys = new java.util.LinkedHashSet<>(); + List results = fileResultMapper.selectList(new LambdaQueryWrapper() + .in(FileResultEntity::getModuleType, moduleTypes) + .in(FileResultEntity::getTaskId, cleanupTaskIds)); + for (FileResultEntity result : results) { + addObjectKey(keys, result.getResultFileUrl()); + } + List jobs = taskFileJobMapper.selectList(new LambdaQueryWrapper() + .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 keys, String value) { + if (value != null && !value.isBlank()) { + keys.add(value.trim()); + } + } + + /** 逐个回收结果对象；单个失败只记日志，不影响其余对象与后续批次。 */ + private void deleteResultObjects(List 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 moduleTypes, List cleanupTaskIds, List collectDataTaskIds) { taskFileJobMapper.delete(new LambdaQueryWrapper() .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），去重并保持稳定顺序。 + * + *

必须覆盖**所有**存放 transient 指针的列：行一旦删除，漏收的指针就再也无法 + * 定位对象，桶里会永久残留孤儿对象。新增写入 payload 指针的列时必须同步加进来。 */ private List collectPayloadPointers(List moduleTypes, List cleanupTaskIds, int max) { java.util.Set pointers = new java.util.LinkedHashSet<>(); @@ -275,8 +375,20 @@ public class ModuleHistoryCleanupService { collectPointer(pointers, scopeState.getParsedPayloadJson(), max); collectPointer(pointers, scopeState.getStateJson(), max); } + List resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper() + .in(TaskResultItemEntity::getModuleType, moduleTypes) + .in(TaskResultItemEntity::getTaskId, cleanupTaskIds)); + for (TaskResultItemEntity resultItem : resultItems) { + collectPointer(pointers, resultItem.getPayloadJson(), max); + } + List resultPayloads = taskResultPayloadMapper.selectList(new LambdaQueryWrapper() + .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); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/usersecret/mapper/UserSecretUsageMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/usersecret/mapper/UserSecretUsageMapper.java index 4b18d6e1..8e020275 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/usersecret/mapper/UserSecretUsageMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/usersecret/mapper/UserSecretUsageMapper.java @@ -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 @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); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/usersecret/service/UserSecretUsageRetentionService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/usersecret/service/UserSecretUsageRetentionService.java new file mode 100644 index 00000000..594ddd00 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/usersecret/service/UserSecretUsageRetentionService.java @@ -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）的保留期清理。 + * + *

每次调用按「用户 × 模块 × 业务日」upsert 累加，只增不删；量级是 + * 用户数 × 模块数 × 天数，单行很小但长期累积无上限。 + * + *

保留期取 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); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryExpiredCleanupService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryExpiredCleanupService.java new file mode 100644 index 00000000..7a142b1d --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryExpiredCleanupService.java @@ -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）的过期行清理。 + * + *

{@link ZiniaoMemoryStoreService#deleteExpired} 早就写好了，但**全库没有任何调用方**—— + * 属于典型的"实现了却没接线"：过期行只有读取命中时才会被顺手删掉一行， + * 店铺下线、改名后遗留的 key 行永远不会有人再读到，于是永久残留。 + * + *

这里把它挂上定时任务。单独成类而不是直接给 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); + } + } +} diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 64b39cf9..2eecbe7d 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -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} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/brand/service/BrandTaskRetentionServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/brand/service/BrandTaskRetentionServiceTest.java new file mode 100644 index 00000000..51ff5b36 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/brand/service/BrandTaskRetentionServiceTest.java @@ -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 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()); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/devicelog/service/DeviceLogRetentionServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/devicelog/service/DeviceLogRetentionServiceTest.java new file mode 100644 index 00000000..f3f195e0 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/devicelog/service/DeviceLogRetentionServiceTest.java @@ -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 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 cutoff = ArgumentCaptor.forClass(LocalDate.class); + verify(fileMapper).deleteOlderThanBatch(cutoff.capture(), anyInt()); + assertEquals(LocalDate.now().minusDays(6), cutoff.getValue(), "未配置时回退默认 7 天"); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunRetentionServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunRetentionServiceTest.java new file mode 100644 index 00000000..1e086e06 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/pricetrack/service/PriceTrackLoopRunRetentionServiceTest.java @@ -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()); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishTaskRetentionServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishTaskRetentionServiceTest.java new file mode 100644 index 00000000..edac5858 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/publish/service/PublishTaskRetentionServiceTest.java @@ -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 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()); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopduplicatecheck/service/ShopDataDuplicateScanRetentionServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopduplicatecheck/service/ShopDataDuplicateScanRetentionServiceTest.java new file mode 100644 index 00000000..d7877dbf --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopduplicatecheck/service/ShopDataDuplicateScanRetentionServiceTest.java @@ -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 protectLine = ArgumentCaptor.forClass(Long.class); + verify(scanMapper).deleteOlderThanBatch(any(LocalDateTime.class), protectLine.capture(), anyInt()); + assertTrue(protectLine.getValue() == Long.MAX_VALUE, "保护线为 Long.MAX_VALUE，等于不额外设限"); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupKeysetPaginationTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupKeysetPaginationTest.java index 142f12f4..f822a842 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupKeysetPaginationTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupKeysetPaginationTest.java @@ -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) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupLoggingTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupLoggingTest.java index bcf579e3..a17aa2b3 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupLoggingTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupLoggingTest.java @@ -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) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java index d1a2b410..f10013b4 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/ModuleHistoryCleanupPayloadCleanupTest.java @@ -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 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、不删除对象； diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/usersecret/service/UserSecretUsageRetentionServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/usersecret/service/UserSecretUsageRetentionServiceTest.java new file mode 100644 index 00000000..f8b9294c --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/usersecret/service/UserSecretUsageRetentionServiceTest.java @@ -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()); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryExpiredCleanupServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryExpiredCleanupServiceTest.java new file mode 100644 index 00000000..5936d03c --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoMemoryExpiredCleanupServiceTest.java @@ -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()); + } +}