From 3adb42c538bb9555544c31ac7df56f157f04aac5 Mon Sep 17 00:00:00 2001 From: root Date: Tue, 15 Sep 2026 16:15:41 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AE=A2=E9=98=85=E5=88=B7=E6=96=B0=E6=94=B9?= =?UTF-8?q?=E4=B8=BA=E6=AF=8F=E8=B4=A6=E5=8F=B7=E9=94=99=E5=B3=B0=EF=BC=8C?= =?UTF-8?q?=E4=BF=9D=E8=AF=81=2024=20=E5=B0=8F=E6=97=B6=E5=86=85=E5=90=84?= =?UTF-8?q?=E5=88=B7=E4=B8=80=E6=AC=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原先每 24 小时一次性刷新全部子账号：12 个账号在 16 秒内背靠背发完 24 个请求，且 @Scheduled(fixedRate) 未设 initialDelay，首次触发即立刻 执行，导致每次重启都重刷一遍全部账号——上游看到的请求密度实际取决于 部署频率。请求头也只带 HttpClient 默认值，上游收到的是 "Apache-HttpClient/5.6.4 (Java/25.0.4)"。 改为高频轻量 tick（5 分钟，延迟 2 分钟启动），只刷新已到期账号，单轮 最多 2 个。24 小时保证由三层构成：成功后按「成功时刻 + 窗口 − 一个 tick」 推进相位，使分散效果自我维持；到期时刻持久化在数据库，停机期间错过的 账号恢复后立即到期；兜底把被写到超过一个窗口之后的计划判为损坏并立即 刷新。失败用固定间隔重试（数据库没有连续失败次数字段，做不出真正的指数 退避），重试排在将来时被尊重，避免故障账号每个 tick 重打上游。从未成功过 的账号按 tick 间隔尽快刷，已有缓存的按窗口分散。超过两倍窗口仍未成功的 账号推送告警。 请求头按账号与订阅格式确定性地伪装成 mihomo/clash-verge/v2rayN 等真实 客户端身份；同一账号固定用同一身份，逐请求更换反而更像机器。实现前已 实测确认显式设置 Accept-Encoding 不会破坏 HttpClient5 的透明解压。注意 这只能伪装到 HTTP 头，TLS 指纹仍与真实客户端不同。 新增 next_refresh_at、last_success_epoch 两列，均为 Epoch 毫秒整数：SQLite 的 CURRENT_TIMESTAMP 写 UTC 文本而被 JDBC 按本地时区解释，实测偏差 8 小时， 不能用于时间比较。迁移为纯新增列，旧二进制忽略它们，故回滚程序无需回滚 数据库；回填 last_success_epoch 必须执行，否则既有账号会被当成从未成功过 而集中补刷，正好复现本次要消除的爆发。 同时移除 LocalService 的 24 小时全量入口与 refreshAll()（它们正是爆发式 写法），并将调度线程池由默认 1 调到 2，避免刷新阻塞连接自检。 测试 439 通过，指令覆盖率 82.1%。 --- scripts/migrate_subscription_refresh.sh | 29 ++ scripts/migrate_subscription_refresh.sql | 26 + .../lionwebsite/Configuration/CustomBean.java | 23 + .../Controller/PersonalController.java | 2 +- .../lionwebsite/Dao/normal/SubMapper.java | 14 +- .../Domain/SubscriptionAccount.java | 10 +- .../lionwebsite/Service/LocalService.java | 116 +---- .../lion/lionwebsite/Service/SubService.java | 6 +- .../Service/SubscriptionRefreshPlanner.java | 225 ++++++++ .../Service/SubscriptionRefreshScheduler.java | 240 +++++++++ .../Service/SubscriptionRefreshService.java | 38 +- .../Util/SubscriptionClientProfile.java | 65 +++ src/main/resources/application.yaml | 19 + .../Controller/PersonalControllerTest.java | 6 +- .../lionwebsite/Service/LocalServiceTest.java | 40 +- .../lionwebsite/Service/SubServiceTest.java | 9 +- .../Service/SubscriptionFilteringTest.java | 68 +-- .../SubscriptionRefreshPlannerTest.java | 338 ++++++++++++ .../SubscriptionRefreshSchedulerTest.java | 481 ++++++++++++++++++ .../SubscriptionRefreshServiceTest.java | 20 +- .../SubscriptionRefreshWiringTest.java | 97 ++++ ...ubscriptionStandbySnapshotServiceTest.java | 9 +- .../Util/SubscriptionClientProfileTest.java | 94 ++++ 23 files changed, 1776 insertions(+), 199 deletions(-) create mode 100755 scripts/migrate_subscription_refresh.sh create mode 100644 scripts/migrate_subscription_refresh.sql create mode 100644 src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshPlanner.java create mode 100644 src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshScheduler.java create mode 100644 src/main/java/com/lion/lionwebsite/Util/SubscriptionClientProfile.java create mode 100644 src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshPlannerTest.java create mode 100644 src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshSchedulerTest.java create mode 100644 src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshWiringTest.java create mode 100644 src/test/java/com/lion/lionwebsite/Util/SubscriptionClientProfileTest.java diff --git a/scripts/migrate_subscription_refresh.sh b/scripts/migrate_subscription_refresh.sh new file mode 100755 index 0000000..1d9876a --- /dev/null +++ b/scripts/migrate_subscription_refresh.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +set -euo pipefail + +db_path="${1:-LionWebsite.db}" + +if [[ ! -f "$db_path" ]]; then + echo "数据库不存在: $db_path" >&2 + exit 2 +fi + +if command -v systemctl >/dev/null 2>&1 && systemctl is-active --quiet lionwebsite 2>/dev/null; then + echo "检测到 lionwebsite.service 仍在运行；请先停止服务再迁移，避免写冲突。" >&2 + exit 3 +fi + +if pgrep -f 'java .*lionwebsite\.jar' >/dev/null 2>&1; then + echo "检测到主站进程仍在运行；请先停止服务再迁移，避免写冲突。" >&2 + exit 3 +fi + +backup="${db_path}.before-refresh-schedule.$(date -u +%Y%m%dT%H%M%SZ)" +cp "$db_path" "$backup" +echo "已备份: $backup" + +sqlite3 "$db_path" < "$(dirname "$0")/migrate_subscription_refresh.sql" + +echo "--- 校验 ---" +sqlite3 "$db_path" "select count(*) as accounts, sum(next_refresh_at is null) as missing_schedule from subscription_account;" +echo "订阅刷新计划迁移完成: $db_path" diff --git a/scripts/migrate_subscription_refresh.sql b/scripts/migrate_subscription_refresh.sql new file mode 100644 index 0000000..7c53b49 --- /dev/null +++ b/scripts/migrate_subscription_refresh.sql @@ -0,0 +1,26 @@ +-- 为每账号分散刷新增加计划字段。 +-- 纯新增列：旧版本二进制忽略这两列，因此回滚程序时不必回滚数据库。 +BEGIN; + +-- next_refresh_at：该子账号下一次应刷新的时刻，Epoch 毫秒。 +-- 用整数存时刻，避免 SQLite 文本时间戳被按本地时区解释（实证存在 8 小时偏差）。 +-- last_success_epoch：最近一次成功刷新的真实时刻，Epoch 毫秒。 +-- 既有的 last_success_at 保持原样，仅供页面展示，不参与调度判断。 +ALTER TABLE subscription_account ADD COLUMN next_refresh_at INTEGER; +ALTER TABLE subscription_account ADD COLUMN last_success_epoch INTEGER; + +-- 回填 last_success_epoch：SQLite 的 strftime 把文本时间按 UTC 解释，而 +-- CURRENT_TIMESTAMP 写入的正是 UTC，因此这里能得到正确时刻（JDBC 读取则会偏 8 小时）。 +-- 必须回填：否则所有既有账号都会被当成「从未成功过」，迁移后立刻集中补刷一遍， +-- 正好复现本次改造要消除的爆发。 +UPDATE subscription_account +SET last_success_epoch = CAST(strftime('%s', last_success_at) AS INTEGER) * 1000 +WHERE last_success_epoch IS NULL AND last_success_at IS NOT NULL; + +-- 已有账号在窗口内按 id 错开，避免迁移后同一 tick 集中开火。 +-- 以迁移时刻为基准按 37 分钟步长铺开（12 个账号约 7.4 小时排完）。 +UPDATE subscription_account +SET next_refresh_at = (CAST(strftime('%s', 'now') AS INTEGER) * 1000) + (id * 37 * 60 * 1000) +WHERE next_refresh_at IS NULL; + +COMMIT; diff --git a/src/main/java/com/lion/lionwebsite/Configuration/CustomBean.java b/src/main/java/com/lion/lionwebsite/Configuration/CustomBean.java index 0f48f4c..ad85b30 100644 --- a/src/main/java/com/lion/lionwebsite/Configuration/CustomBean.java +++ b/src/main/java/com/lion/lionwebsite/Configuration/CustomBean.java @@ -2,6 +2,7 @@ package com.lion.lionwebsite.Configuration; import com.lion.lionwebsite.Domain.*; import com.lion.lionwebsite.Message.*; +import com.lion.lionwebsite.Service.SubscriptionRefreshPlanner; import com.lion.lionwebsite.Util.GalleryUtil; import com.pengrad.telegrambot.TelegramBot; import com.pengrad.telegrambot.model.*; @@ -14,6 +15,10 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import java.time.Clock; +import java.time.Duration; +import java.util.Random; + @Configuration @RegisterReflectionForBinding(classes = {CustomConfiguration.class, GidToKey.class, ImageKeyCache.class, GalleryForQuery.class, Gallery.class, GalleryTask.class, HikariConfig.class, @@ -41,4 +46,22 @@ public class CustomBean { public TelegramBot getTelegramBot(){ return new TelegramBot(botToken); } + + /** + * 订阅分散刷新的排程器。 + * + *

放在这里是为了让窗口、最小间隔、tick 周期与重试间隔集中由配置注入， + * 同时保持 {@link SubscriptionRefreshPlanner} 本身是可直接构造的纯对象 + * （便于用固定时钟与固定随机种子做确定性单元测试）。 + */ + @Bean + public SubscriptionRefreshPlanner subscriptionRefreshPlanner( + @Value("${subscription.refresh.window-hours:24}") long windowHours, + @Value("${subscription.refresh.min-gap-minutes:60}") long minGapMinutes, + @Value("${subscription.refresh.tick-interval-ms:300000}") long tickIntervalMs, + @Value("${subscription.refresh.retry-delay-minutes:60}") long retryDelayMinutes) { + return new SubscriptionRefreshPlanner(Clock.systemDefaultZone(), new Random(), + Duration.ofHours(windowHours), Duration.ofMinutes(minGapMinutes), + Duration.ofMillis(tickIntervalMs), Duration.ofMinutes(retryDelayMinutes)); + } } diff --git a/src/main/java/com/lion/lionwebsite/Controller/PersonalController.java b/src/main/java/com/lion/lionwebsite/Controller/PersonalController.java index c4dce4b..0d977a4 100644 --- a/src/main/java/com/lion/lionwebsite/Controller/PersonalController.java +++ b/src/main/java/com/lion/lionwebsite/Controller/PersonalController.java @@ -74,7 +74,7 @@ public class PersonalController { @PostMapping("/updateSub") public String updateSub() throws IOException { Response response = Response.generateResponse(); - if(localService.updateSub(true)) + if(localService.updateSub()) response.success(); else response.failure(); diff --git a/src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java b/src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java index 1bc52f0..7a65210 100644 --- a/src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java +++ b/src/main/java/com/lion/lionwebsite/Dao/normal/SubMapper.java @@ -13,10 +13,10 @@ public interface SubMapper { @Options(useGeneratedKeys = true, keyProperty = "id") void insertSubscriptionAccount(SubscriptionAccount account); - @Select("select id, name, upstream_key as upstreamKey, filter_high_multiplier as filterHighMultiplier, enabled, last_success_at as lastSuccessAt, last_error as lastError, created_at as createdAt, updated_at as updatedAt, (select count(*) from sub_bind sb where sb.subscription_account_id = sa.id) as boundUserCount from subscription_account sa order by id") + @Select("select id, name, upstream_key as upstreamKey, filter_high_multiplier as filterHighMultiplier, enabled, last_success_at as lastSuccessAt, last_error as lastError, created_at as createdAt, updated_at as updatedAt, next_refresh_at as nextRefreshAt, last_success_epoch as lastSuccessEpoch, (select count(*) from sub_bind sb where sb.subscription_account_id = sa.id) as boundUserCount from subscription_account sa order by id") ArrayList selectAllSubscriptionAccounts(); - @Select("select id, name, upstream_key as upstreamKey, filter_high_multiplier as filterHighMultiplier, enabled, last_success_at as lastSuccessAt, last_error as lastError, created_at as createdAt, updated_at as updatedAt, (select count(*) from sub_bind sb where sb.subscription_account_id = sa.id) as boundUserCount from subscription_account sa where id=#{id}") + @Select("select id, name, upstream_key as upstreamKey, filter_high_multiplier as filterHighMultiplier, enabled, last_success_at as lastSuccessAt, last_error as lastError, created_at as createdAt, updated_at as updatedAt, next_refresh_at as nextRefreshAt, last_success_epoch as lastSuccessEpoch, (select count(*) from sub_bind sb where sb.subscription_account_id = sa.id) as boundUserCount from subscription_account sa where id=#{id}") SubscriptionAccount selectSubscriptionAccount(Integer id); @Select("select count(*) from subscription_account where name=#{name}") @@ -28,12 +28,18 @@ public interface SubMapper { @Update("update subscription_account set name=#{name}, upstream_key=#{upstreamKey}, filter_high_multiplier=#{filterHighMultiplier}, enabled=#{enabled}, updated_at=CURRENT_TIMESTAMP, last_error=null where id=#{id}") void updateSubscriptionAccount(SubscriptionAccount account); - @Update("update subscription_account set last_success_at=CURRENT_TIMESTAMP, last_error=null, updated_at=CURRENT_TIMESTAMP where id=#{id}") - void markSubscriptionRefreshSuccess(Integer id); + // last_success_at 仍写 SQLite 的 UTC 文本时间，仅供页面显示； + // last_success_epoch 是调度判断用的权威时刻，两者口径不同，不要互相推导。 + @Update("update subscription_account set last_success_at=CURRENT_TIMESTAMP, last_success_epoch=#{epoch}, last_error=null, updated_at=CURRENT_TIMESTAMP where id=#{id}") + void markSubscriptionRefreshSuccess(@Param("id") Integer id, @Param("epoch") long epoch); @Update("update subscription_account set last_error=#{error}, updated_at=CURRENT_TIMESTAMP where id=#{id}") void markSubscriptionRefreshFailure(@Param("id") Integer id, @Param("error") String error); + /** 排定下一次刷新时刻；失败重试与次日分槽都走这里。 */ + @Update("update subscription_account set next_refresh_at=#{nextRefreshAt}, updated_at=CURRENT_TIMESTAMP where id=#{id}") + void updateNextRefreshAt(@Param("id") Integer id, @Param("nextRefreshAt") long nextRefreshAt); + @Delete("delete from subscription_account where id=#{id}") void deleteSubscriptionAccount(Integer id); diff --git a/src/main/java/com/lion/lionwebsite/Domain/SubscriptionAccount.java b/src/main/java/com/lion/lionwebsite/Domain/SubscriptionAccount.java index 85470c3..6dbdcf2 100644 --- a/src/main/java/com/lion/lionwebsite/Domain/SubscriptionAccount.java +++ b/src/main/java/com/lion/lionwebsite/Domain/SubscriptionAccount.java @@ -1,6 +1,5 @@ package com.lion.lionwebsite.Domain; -import lombok.AllArgsConstructor; import lombok.Data; import lombok.NoArgsConstructor; @@ -8,7 +7,6 @@ import java.util.Date; @Data @NoArgsConstructor -@AllArgsConstructor public class SubscriptionAccount { private Integer id; private String name; @@ -22,4 +20,12 @@ public class SubscriptionAccount { private Integer boundUserCount; private String v2Url; private String clashUrl; + /** + * 下一次应刷新的时刻（Epoch 毫秒），由分散调度器排定。 + * 刻意用整数而非 DATETIME：SQLite 文本时间戳会被 JDBC 按本地时区解释， + * 实测偏差 8 小时，不能用于时间比较。 + */ + private Long nextRefreshAt; + /** 最近一次成功刷新的真实时刻（Epoch 毫秒），用于「24 小时内必刷一次」的判断与陈旧告警。 */ + private Long lastSuccessEpoch; } diff --git a/src/main/java/com/lion/lionwebsite/Service/LocalService.java b/src/main/java/com/lion/lionwebsite/Service/LocalService.java index cb2aa23..b11cb35 100644 --- a/src/main/java/com/lion/lionwebsite/Service/LocalService.java +++ b/src/main/java/com/lion/lionwebsite/Service/LocalService.java @@ -53,7 +53,7 @@ public class LocalService{ final RemoteService remoteService; - final SubscriptionRefreshService subscriptionRefreshService; + final SubscriptionRefreshScheduler subscriptionRefreshScheduler; /** * 检查连接是否有效，如果无效自动重连 @@ -104,115 +104,23 @@ public class LocalService{ } /** - * 定时更新订阅 - * @throws IOException 下载以及保存异常 + * 手动「更新订阅」：立即刷新全部启用账号，供管理页按钮调用。 + * + *

定时刷新已移交 {@link SubscriptionRefreshScheduler}。原先这里还有一个 + * {@code @Scheduled(fixedRate = 86400000)} 的定时入口，它会一次性刷新全部账号 + * （十几个账号在十几秒内打满），且因为没有 initialDelay，每次重启都会立刻重刷一遍， + * 使上游看到的请求密度取决于部署频率。该入口已移除。 + * + *

不再保留 {@code isManual} 参数：定时路径已不存在，留着它只会让人以为 + * 「传 false 就是不刷新」，而实际语义是「什么都不做却报成功」。 */ - @Scheduled(fixedRate = 86400000) - public void updateSubScheduler() throws IOException { - updateSub(false); - } - - /** - * 更新订阅链接的实际方法 - */ - public boolean updateSub(boolean isManual) throws IOException { - // 手动和定时入口均刷新全部启用子账号；全部成功后更新原有“上次更新时间”。 - boolean success = subscriptionRefreshService.refreshAll(); + public boolean updateSub() { + boolean success = subscriptionRefreshScheduler.refreshAllNow(); remoteService.requestSubscriptionSync(); if (success) configurationMapper.updateConfiguration(CustomConfiguration.LAST_UPDATE_SUB_TIME, CustomUtil.dateTimeFormatter().format(LocalDateTime.now())); return success; - /* - DateTimeFormatter dateTimeFormatter = CustomUtil.dateTimeFormatter(); - CustomConfiguration customConfiguration = configurationMapper.selectConfiguration(CustomConfiguration.LAST_UPDATE_SUB_TIME); - - //如果不是手动，则判断更新间隔是否满足，不满足则取消更新 - if (!isManual) { - LocalDateTime lastUpdate = LocalDateTime.parse(customConfiguration.getValue(), CustomUtil.dateTimeFormatter()); - LocalDateTime now = LocalDateTime.now(); - now = now.plusHours(-3); - if(now.isBefore(lastUpdate)) - return false; - } - - File DouNaiClashFile = new File("sub/DouNaiClash.txt"); - File DouNaiV2rayFile = new File("sub/DouNaiV2ray.txt"); - File directory = new File("sub"); - - if(!directory.isDirectory()) - Files.createDirectory(Paths.get("sub")); - - List DouNaiClash_profile; - - //下载豆奶v2ray订阅 - try(FileWriter writer = new FileWriter(DouNaiV2rayFile)) { - String DouNaiV2rayRaw = Get(DouNaiV2ray).getFirst(); - String[] v2rayPlain = new String(Base64.getDecoder().decode(DouNaiV2rayRaw)).split("\n"); - StringBuilder stringBuilder = new StringBuilder(); - Pattern pattern = Pattern.compile("-?\\d+(\\.\\d+)?"); - - //过滤高倍率节点 - for(String node: v2rayPlain){ - String name = URLDecoder.decode(node.split("#")[1], StandardCharsets.UTF_8); - if(name.contains("流量")){ - Matcher matcher = pattern.matcher(name.substring(name.indexOf("(") + 1, name.indexOf(")"))); - - if (matcher.find()) { - // 将匹配到的数字添加到列表中 - float ratio = Float.parseFloat(matcher.group()); - if(ratio <= 2) { - stringBuilder.append(node).append("\n"); - continue; - } - } - stringBuilder.append(node).append("\n"); - } - else{ - stringBuilder.append(node).append("\n"); - } - } - writer.write(new String(Base64.getEncoder().encode(stringBuilder.toString().getBytes(StandardCharsets.UTF_8)))); - - log.info("load DouNai v2ray complete"); - }catch (IOException e){ - log.error("load DouNai v2ray failure", e); - } - - //下载豆奶clash订阅 - try(FileWriter writer = new FileWriter(DouNaiClashFile)) { - DouNaiClash_profile = Get(DouNaiClash); - //过滤高倍率节点 - ArrayList clashProcessed = new ArrayList<>(); - boolean isProxies = false; - boolean skip = false; - for(String line: DouNaiClash_profile){ - if(line.equals("proxies:")) - isProxies = true; - else if(line.equals("proxy-groups:") && isProxies) - isProxies = false; - - if(isProxies) { - if (line.contains("name")) - skip = line.contains("流量"); - if (!skip) - clashProcessed.add(line); - } - else - if (!line.contains("流量")) - clashProcessed.add(line); - } - - for(String line: clashProcessed) - writer.write(line + "\n"); - log.info("load DouNai clash complete"); - }catch (IOException e){ - log.error("load DouNai clash failure", e); - } - - configurationMapper.updateConfiguration(CustomConfiguration.LAST_UPDATE_SUB_TIME, dateTimeFormatter.format(LocalDateTime.now())); - return true; - */ } /** diff --git a/src/main/java/com/lion/lionwebsite/Service/SubService.java b/src/main/java/com/lion/lionwebsite/Service/SubService.java index 215ee9c..9ffa833 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubService.java @@ -46,7 +46,11 @@ public class SubService { return response.failure("名称和上游 key 不能为空").toJSONString(); if (subMapper.countSubscriptionAccountName(name.trim()) > 0 || subMapper.countSubscriptionAccountKey(upstreamKey.trim()) > 0) return response.failure("名称或上游 key 已存在").toJSONString(); - account = new SubscriptionAccount(null, name.trim(), upstreamKey.trim(), filterHighMultiplier, enabled, null, null, null, null, 0, null, null); + account = new SubscriptionAccount(); + account.setName(name.trim()); + account.setUpstreamKey(upstreamKey.trim()); + account.setFilterHighMultiplier(filterHighMultiplier); + account.setEnabled(enabled); subMapper.insertSubscriptionAccount(account); } finally { lock.unlock(); diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshPlanner.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshPlanner.java new file mode 100644 index 0000000..0196cd7 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshPlanner.java @@ -0,0 +1,225 @@ +package com.lion.lionwebsite.Service; + +import java.time.Clock; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Random; + +/** + * 为一批子账号排定刷新时刻。 + * + *

目标：每个账号每 24 小时内至少刷新一次，且各账号的时刻互相错开， + * 不出现「一次全量、十几个账号在十几秒内打完」的爆发式请求。 + * + *

排程模型是「错峰 + 固定相位」： + *

    + *
  1. 错峰：首次排程把账号随机铺满一个窗口，两两之间至少隔开 {@code minGap}；
    2. + *
  2. 固定相位：账号每次成功后按窗口长度（略减一个 tick）推进， + * 因此各自的相位保持不变，分散效果自我维持，不需要每天重新洗牌；
    3. + *
  3. 兜底：任何情况下下一次刷新都不会晚于「上次成功 + 窗口」， + * 把「24 小时内必然刷新一次」从概率保证变成确定保证。
    4. + *
+ * + *

本类是纯函数式设计（注入 {@link Clock} 与带种子的 {@link Random}）， + * 不触碰数据库与网络，因此可以直接做确定性单元测试。 + */ +public class SubscriptionRefreshPlanner { + + private final Clock clock; + private final Random random; + private final Duration window; + private final Duration minGap; + /** 一个调度周期的最大延迟；用它给窗口留余量，使实际间隔严格落在窗口内。 */ + private final Duration tickInterval; + /** 刷新失败后的重试间隔。 */ + private final Duration retryDelay; + + /** + * @param clock 时间源，测试可注入固定时钟 + * @param random 随机源，测试可注入固定种子以复现排程 + * @param window 刷新窗口，即「24 小时内至少一次」中的 24 小时 + * @param minGap 相邻两个账号之间的最小间隔 + * @param tickInterval 调度周期，用于给窗口留出余量 + * @param retryDelay 失败后的重试间隔 + */ + public SubscriptionRefreshPlanner(Clock clock, Random random, Duration window, Duration minGap, + Duration tickInterval, Duration retryDelay) { + if (window.isZero() || window.isNegative()) + throw new IllegalArgumentException("刷新窗口必须为正数"); + if (minGap.isNegative()) + throw new IllegalArgumentException("最小间隔不能为负数"); + if (tickInterval.isNegative()) + throw new IllegalArgumentException("调度周期不能为负数"); + if (retryDelay.isZero() || retryDelay.isNegative()) + throw new IllegalArgumentException("重试间隔必须为正数"); + if (retryDelay.compareTo(window) > 0) + throw new IllegalArgumentException("重试间隔不能大于刷新窗口，否则无法保证窗口内重试"); + this.clock = clock; + this.random = random; + this.window = window; + this.minGap = minGap; + this.tickInterval = tickInterval; + this.retryDelay = retryDelay; + } + + /** 便捷构造：重试间隔默认取 1 小时。 */ + public SubscriptionRefreshPlanner(Clock clock, Random random, Duration window, Duration minGap, + Duration tickInterval) { + this(clock, random, window, minGap, tickInterval, Duration.ofHours(1)); + } + + /** + * 账号数与最小间隔是否放得下。 + * + *

注意这里刻意只用于「提示」而不是阻断启动：账号数是由用户在管理页决定的， + * 若因为它超限就让整个应用起不来，等于把配置细节变成一次线上故障。 + * 放不下时 {@link #initialSchedule} 仍会把账号铺满整个窗口（只是间隔小于期望值）， + * 结果是错峰效果变弱，而不是退回串行爆发。 + */ + public boolean isCapacitySufficient(int accountCount) { + if (accountCount <= 1) + return true; + return minGap.toMillis() * (accountCount - 1) <= window.toMillis(); + } + + /** 容量不足时的告警文案，供调用方记录日志。 */ + public String capacityMessage(int accountCount) { + long required = minGap.toMillis() * (accountCount - 1); + return String.format( + "刷新窗口放不下最小间隔：账号 %d 个、最小间隔 %d 分钟，需要 %d 分钟，但窗口只有 %d 分钟；" + + "实际仍会铺满窗口，只是账号间间隔会小于期望值。可调小 subscription.refresh.min-gap-minutes。", + accountCount, minGap.toMinutes(), + Duration.ofMillis(required).toMinutes(), window.toMinutes()); + } + + /** + * 为一批「已成功刷新过」的账号铺开计划，返回顺序与 accountIds 一致。 + * + *

已成功过的账号内容仍然可用，因此可以安心分散到整个窗口， + * 不会因为排在 20 小时后而让用户拿不到订阅。 + */ + public List initialSchedule(List accountIds) { + List result = new ArrayList<>(accountIds.size()); + if (accountIds.isEmpty()) + return result; + + long now = clock.millis(); + long windowMillis = window.toMillis(); + int count = accountIds.size(); + long slotWidth = windowMillis / count; + long jitterCap = Math.max(1, Math.min(slotWidth, Math.max(1, slotWidth - minGap.toMillis()))); + + List slots = new ArrayList<>(count); + for (int i = 0; i < count; i++) + slots.add(i); + Collections.shuffle(slots, random); + + long[] planned = new long[count]; + for (int accountIndex = 0; accountIndex < count; accountIndex++) { + long offset = slots.get(accountIndex) * slotWidth + (long) (random.nextDouble() * jitterCap); + planned[accountIndex] = now + Math.min(offset, windowMillis - 1); + } + for (int i = 0; i < count; i++) + result.add(planned[i]); + return result; + } + + /** + * 为尚未成功过的账号排程：尽快刷，但互相错开。 + * + *

这类账号（新建、或上游 key 刚改过）可能根本没有可用缓存，用户立刻请求订阅 + * 就会拿不到内容，因此不能像常规账号那样分散到整个窗口。这里把它们按 + * {@code spacing} 依次排开，配合调度器的单轮上限逐批刷新， + * 既尽早补齐，又避免一批新账号同时打上游。 + * + * @param accountIds 账号 ID，顺序即刷新优先级 + * @param spacing 相邻两个账号的间隔（通常取一个 tick） + */ + public List coldStartSchedule(List accountIds, Duration spacing) { + List result = new ArrayList<>(accountIds.size()); + long now = clock.millis(); + long step = Math.max(1, spacing.toMillis()); + for (int i = 0; i < accountIds.size(); i++) + result.add(now + i * step); + return result; + } + + /** + * 已排定的时刻是否已到期。 + * + *

直接以排定时刻为准，这样分散效果与失败退避都得以保持。 + * 只额外兜住一种异常：计划被写到超过一个窗口之后。这不可能是本调度器产生的 + * 排程（成功时排在窗口减一个 tick 之后，失败时排在 4 小时以内）， + * 按「计划损坏」处理并立即到期，从而保证 24 小时内必定刷新。 + * + *

尚未排程（{@code scheduled == null}）视为立即到期，由调用方先行铺开计划。 + */ + public boolean isDue(Long scheduled, long now) { + if (scheduled == null) + return true; + if (scheduled - now > window.toMillis()) + return true; + return scheduled <= now; + } + + /** + * 成功之后的下一次刷新时刻。 + * + *

只推进「窗口 - 一个调度周期」，使 tick 即使晚触发也不会突破 24 小时； + * 从上次成功的时刻起算，保证相邻两次的间隔严格小于窗口。 + */ + public long nextAfterSuccess(long successMillis) { + long advance = Math.max(1, window.minus(tickInterval).toMillis()); + return successMillis + advance; + } + + /** + * 失败之后的重试时刻。 + * + *

用固定的重试间隔而不是「失败次数 × 指数退避」：刷新成功会清空 {@code last_error}， + * 数据库里没有「连续失败次数」这一栏，退避指数实际上只能取到 0 或 1， + * 写出来会让人误以为有指数退避而其实没有。固定的重试间隔行为可预期， + * 且远小于窗口，不会威胁「24 小时内尝试刷新一次」。 + * + *

唯一需要裁剪的情况：若「上次成功 + 窗口」还没到，则重试不得晚于它； + * 已经越过该点时保留重试时刻本身——重复请求并不会提高成功率， + * 反过来在每个 tick 重打上游才是真正的问题，这种情况交由陈旧告警处理。 + * + * @param now 当前时刻 + * @param lastSuccess 上次成功时刻；未知时传 null，不做裁剪 + */ + public long nextAfterFailure(long now, Long lastSuccess) { + long candidate = now + retryDelay.toMillis(); + if (lastSuccess == null) + return candidate; + long deadline = lastSuccess + window.toMillis(); + return deadline > now ? Math.min(candidate, deadline) : candidate; + } + + /** 是否已越过兜底线（距上次成功已达一个窗口），用于诊断与告警判定。 */ + public boolean isOverdue(Long lastSuccess, long now) { + return lastSuccess != null && now - lastSuccess >= window.toMillis(); + } + + private long now() { + return clock.millis(); + } + + public Clock clock() { + return clock; + } + + public Duration window() { + return window; + } + + public Duration minGap() { + return minGap; + } + + public Duration tickInterval() { + return tickInterval; + } +} diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshScheduler.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshScheduler.java new file mode 100644 index 0000000..f5c827b --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshScheduler.java @@ -0,0 +1,240 @@ +package com.lion.lionwebsite.Service; + +import com.lion.lionwebsite.Dao.normal.SubMapper; +import com.lion.lionwebsite.Domain.SubscriptionAccount; +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.util.ArrayList; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** + * 分散刷新各子账号的订阅。 + * + *

取代原先「每 24 小时一次全量」的调度。旧写法有两个问题：所有账号在十几秒内 + * 背靠背打完，且 {@code @Scheduled(fixedRate = 86400000)} 没有 initialDelay， + * 首次触发即立刻执行——于是每次重启都会重刷一遍全部账号，上游看到的请求密度取决于 + * 部署频率，而不是每天一次。 + * + *

现在改为高频轻量 tick，每轮只刷新已经到期的账号。「24 小时内每个账号至少 + * 更新一次」由三层保证： + *

    + *
  1. 成功后排定「成功时刻 + 窗口 - 一个 tick」，相位自我维持，分散效果不会因 + * 重启或重排而退化；
    2. + *
  2. 到期时刻存在数据库里，停机期间错过计划的账号在恢复后立即到期；
    3. + *
  3. 兜底闸：任何排定时刻都被裁剪到「上次成功 + 窗口」以内，越线即立刻刷新。
    4. + *
+ */ +@Service +@Slf4j +public class SubscriptionRefreshScheduler { + + private final SubMapper subMapper; + private final SubscriptionRefreshService refreshService; + private final SubscriptionRefreshPlanner planner; + private final PushService pushService; + + /** 单个 tick 最多刷新几个账号，避免重启后一批到期账号同时开火。 */ + @Value("${subscription.refresh.max-per-tick:2}") + int maxPerTick; + + /** 超过窗口的这个倍数仍未成功即告警，避免静默失败。 */ + @Value("${subscription.refresh.stale-alert-multiplier:2.0}") + double staleAlertMultiplier; + + /** 已告警过的账号，避免每轮重复推送；刷新成功后移除。 */ + private final Set alertedAccounts = new HashSet<>(); + + public SubscriptionRefreshScheduler(SubMapper subMapper, + SubscriptionRefreshService refreshService, + SubscriptionRefreshPlanner planner, + PushService pushService) { + this.subMapper = subMapper; + this.refreshService = refreshService; + this.planner = planner; + this.pushService = pushService; + } + + /** + * 轻量 tick：只处理已到期的账号，且单轮有数量上限。 + * + *

用 fixedDelay 而非 fixedRate，一轮结束再计下一轮，避免上一轮没跑完就叠加触发； + * initialDelay 让重启后的第一轮稍晚开始，先让应用完成启动。 + */ + @Scheduled(fixedDelayString = "${subscription.refresh.tick-interval-ms:300000}", + initialDelayString = "${subscription.refresh.initial-delay-ms:120000}") + public void tick() { + try { + refreshDueAccounts(); + } catch (Exception e) { + // 调度方法抛出异常会导致后续触发被取消，这里必须兜住。 + log.error("订阅分散刷新 tick 失败", e); + } + } + + /** + * 刷新当前已到期的账号。 + * + * @return 本轮实际刷新的账号数 + */ + public int refreshDueAccounts() { + List enabled = enabledAccounts(); + ensureScheduled(enabled); + + long now = planner.clock().millis(); + int refreshed = 0; + for (SubscriptionAccount account : enabled) { + if (refreshed >= maxPerTick) + break; + if (!isDue(account, now)) + continue; + refreshOne(account, now); + refreshed++; + } + reportStale(enabled, now); + return refreshed; + } + + /** + * 手动刷新入口：立即刷新全部启用账号。 + * + *

刻意不受 {@code maxPerTick} 限制——管理页「更新订阅」的语义就是立刻全部刷新。 + * 但成功后同样要重排各自的下一次时刻，否则刚手动刷完会让当天的计划作废， + * 该账号反而可能超过 24 小时没有下次更新。 + */ + public boolean refreshAllNow() { + boolean success = true; + long now = planner.clock().millis(); + for (SubscriptionAccount account : enabledAccounts()) { + if (!refreshOne(account, now)) + success = false; + } + return success; + } + + private List enabledAccounts() { + return subMapper.selectAllSubscriptionAccounts().stream() + .filter(SubscriptionAccount::isEnabled) + .sorted(Comparator.comparing(SubscriptionAccount::getId)) + .toList(); + } + + /** 刷新单个账号并按结果重排下一次时刻。 */ + private boolean refreshOne(SubscriptionAccount account, long now) { + boolean ok = refreshService.refresh(account.getId()); + if (ok) { + long next = planner.nextAfterSuccess(planner.clock().millis()); + subMapper.updateNextRefreshAt(account.getId(), next); + alertedAccounts.remove(account.getId()); + log.info("订阅刷新成功 accountId={} 下次刷新={}", account.getId(), next); + } else { + long next = planner.nextAfterFailure(now, account.getLastSuccessEpoch()); + subMapper.updateNextRefreshAt(account.getId(), next); + log.warn("订阅刷新失败 accountId={} 下次重试={}", account.getId(), next); + } + return ok; + } + + /** + * 账号是否到期。 + * + *

以排定时刻为准；尚未排程或计划明显损坏（超出窗口）时视为立即到期， + * 由 {@link #ensureScheduled} 先铺开或纠正。 + */ + private boolean isDue(SubscriptionAccount account, long now) { + return planner.isDue(account.getNextRefreshAt(), now); + } + + /** + * 为尚无计划的账号补齐计划。 + * + *

按紧迫性分三类，因为它们对「排到多晚」的容忍度不同： + *

+ */ + private void ensureScheduled(List enabled) { + long now = planner.clock().millis(); + List cold = new ArrayList<>(); + List overdue = new ArrayList<>(); + List warm = new ArrayList<>(); + for (SubscriptionAccount account : enabled) { + if (account.getNextRefreshAt() != null) + continue; + if (account.getLastSuccessEpoch() == null) + cold.add(account); + else if (planner.isOverdue(account.getLastSuccessEpoch(), now)) + overdue.add(account); + else + warm.add(account); + } + if (cold.isEmpty() && overdue.isEmpty() && warm.isEmpty()) + return; + + if (!planner.isCapacitySufficient(enabled.size())) + log.warn("{}", planner.capacityMessage(enabled.size())); + + // 超期账号立即到期（计划设为当前时刻），本轮就会被刷新。 + for (SubscriptionAccount account : overdue) + schedule(account, now); + + int index = 0; + List coldPlan = planner.coldStartSchedule( + cold.stream().map(SubscriptionAccount::getId).toList(), planner.tickInterval()); + for (SubscriptionAccount account : cold) + schedule(account, coldPlan.get(index++)); + + List warmIds = warm.stream().map(SubscriptionAccount::getId).toList(); + List warmPlan = planner.initialSchedule(warmIds); + for (int i = 0; i < warm.size(); i++) + schedule(warm.get(i), warmPlan.get(i)); + } + + /** + * 写入新计划，并同步更新内存中的对象。 + * + *

必须同时更新内存副本：否则刚被排到几小时后的账号在本轮仍带着 + * {@code nextRefreshAt == null}，会被 {@link #isDue} 判为立即到期并当场刷新， + * 于是「铺开」在第一个 tick 就失效了（迁移后首批账号会全部集中刷新）。 + */ + private void schedule(SubscriptionAccount account, long nextRefreshAt) { + subMapper.updateNextRefreshAt(account.getId(), nextRefreshAt); + account.setNextRefreshAt(nextRefreshAt); + } + + /** 陈旧账号告警：同一账号在连续陈旧期间只推一次，刷新成功后重置。 */ + private void reportStale(List enabled, long now) { + long threshold = (long) (planner.window().toMillis() * staleAlertMultiplier); + for (SubscriptionAccount account : enabled) { + Long lastSuccess = account.getLastSuccessEpoch(); + if (lastSuccess == null || now - lastSuccess < threshold) + continue; + if (!alertedAccounts.add(account.getId())) + continue; + long hours = (now - lastSuccess) / 3_600_000L; + String message = String.format("订阅账号 %d（%s）已 %d 小时未成功刷新，最近错误：%s", + account.getId(), account.getName(), hours, + account.getLastError() == null ? "无" : account.getLastError()); + log.error("{}", message); + pushService.sendToMe(message); + } + // 已删除或停用的账号不再保留告警状态，避免集合无限增长。 + Set alive = new HashSet<>(); + for (SubscriptionAccount account : enabled) + alive.add(account.getId()); + alertedAccounts.retainAll(alive); + } + + SubscriptionRefreshPlanner planner() { + return planner; + } +} diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java index d18f097..2cb1bae 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java +++ b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java @@ -2,6 +2,7 @@ package com.lion.lionwebsite.Service; import com.lion.lionwebsite.Dao.normal.SubMapper; import com.lion.lionwebsite.Domain.SubscriptionAccount; +import com.lion.lionwebsite.Util.SubscriptionClientProfile; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.hc.client5.http.classic.methods.HttpGet; @@ -18,6 +19,7 @@ import java.io.*; import java.net.URLDecoder; import java.nio.charset.StandardCharsets; import java.nio.file.*; +import java.time.Clock; import java.util.*; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -52,14 +54,8 @@ public class SubscriptionRefreshService { @Value("${subscription.cache-root:sub/accounts}") String cacheRoot; - public boolean refreshAll() { - boolean success = true; - for (SubscriptionAccount account : subMapper.selectAllSubscriptionAccounts()) { - if (account.isEnabled() && !refresh(account.getId())) - success = false; - } - return success; - } + /** 时间源，测试可替换以固定成功时刻。 */ + Clock clock = Clock.systemDefaultZone(); private long refreshSequence; // Accessed only while holding the coordinator write lock. @@ -82,8 +78,10 @@ public class SubscriptionRefreshService { // Network access and parsing never hold the shared subscription lock. try { - String v2 = processV2(firstLine(download(v2Url(account))), account.isFilterHighMultiplier(), highMultiplierThreshold); - List clash = processClash(download(clashUrl(account)), account.isFilterHighMultiplier(), highMultiplierThreshold); + String v2 = processV2(firstLine(download(v2Url(account), SubscriptionClientProfile.forAccount(accountId, false))), + account.isFilterHighMultiplier(), highMultiplierThreshold); + List clash = processClash(download(clashUrl(account), SubscriptionClientProfile.forAccount(accountId, true)), + account.isFilterHighMultiplier(), highMultiplierThreshold); stateLock.lock(); try { if (!isCurrent(account, version)) @@ -92,7 +90,7 @@ public class SubscriptionRefreshService { Files.createDirectories(dir); atomicWrite(dir.resolve("v2ray.txt"), v2.getBytes(StandardCharsets.UTF_8)); atomicWrite(dir.resolve("clash.yaml"), String.join("\n", clash).concat("\n").getBytes(StandardCharsets.UTF_8)); - subMapper.markSubscriptionRefreshSuccess(accountId); + subMapper.markSubscriptionRefreshSuccess(accountId, clock.millis()); return true; } finally { stateLock.unlock(); @@ -219,8 +217,24 @@ public class SubscriptionRefreshService { return matcher.find() && Double.parseDouble(matcher.group(1)) > threshold; } - List download(String url) throws IOException { + /** + * 下载上游订阅正文。 + * + * @param profile 要伪装的客户端身份；为 null 时保留 HttpClient 默认头（仅测试使用）。 + * 生产路径必须传入，否则上游会看到 {@code Apache-HttpClient/... (Java/...)}。 + */ + List download(String url, SubscriptionClientProfile profile) throws IOException { HttpGet get = new HttpGet(url); + if (profile != null) { + get.addHeader("User-Agent", profile.userAgent()); + get.addHeader("Accept", profile.accept()); + get.addHeader("Accept-Language", profile.acceptLanguage()); + // 实测：显式设置 Accept-Encoding 不会破坏 HttpClient5 的透明解压， + // ContentCompressionExec 仍按 Content-Encoding 正确解码。 + get.addHeader("Accept-Encoding", "gzip, deflate"); + // 订阅客户端是长连接复用，且不会发送 Referer。 + get.addHeader("Connection", "keep-alive"); + } try (CloseableHttpResponse response = HTTP_CLIENT.execute(get)) { if (response.getCode() != 200) throw new IOException("上游 HTTP 状态码 " + response.getCode()); diff --git a/src/main/java/com/lion/lionwebsite/Util/SubscriptionClientProfile.java b/src/main/java/com/lion/lionwebsite/Util/SubscriptionClientProfile.java new file mode 100644 index 0000000..565b789 --- /dev/null +++ b/src/main/java/com/lion/lionwebsite/Util/SubscriptionClientProfile.java @@ -0,0 +1,65 @@ +package com.lion.lionwebsite.Util; + +import java.util.List; + +/** + * 抓取上游订阅时伪装的客户端身份。 + * + *

此前请求只带 HttpClient 的默认头，实测上游收到的是 + * {@code User-agent: Apache-HttpClient/5.6.4 (Java/25.0.4)}， + * 对订阅站点而言等同于「这是一个 Java 程序在批量拉取」。 + * 这里按订阅格式给出与真实客户端一致的常见头组合。 + * + *

注意：只能伪装到 HTTP 头这一层。真实 Clash/Mihomo 使用 Go 的 TLS 栈与 + * HTTP/2，TLS 指纹与 Java HttpClient 仍有差异；若上游做主动指纹识别， + * 需要另外引入真实客户端进程，本枚举解决不了。 + */ +public enum SubscriptionClientProfile { + MIHOMO("mihomo/v1.19.2", "application/yaml, */*", "zh-CN,zh;q=0.9"), + CLASH_VERGE("clash-verge/v2.0.3", "application/yaml, */*", "zh-CN,zh;q=0.9"), + CLASH_WINDOWS("ClashforWindows/0.20.39", "*/*", "zh-CN,zh;q=0.9"), + V2RAYN("v2rayN/6.45", "*/*", "zh-CN,zh;q=0.9"), + V2RAYNG("v2rayNG/1.9.16", "*/*", "zh-CN,zh;q=0.9"), + SHADOWROCKET("Shadowrocket/2.2.39", "*/*", "zh-CN,zh;q=0.9"); + + static final List CLASH_PROFILES = List.of(MIHOMO, CLASH_VERGE, CLASH_WINDOWS); + static final List V2_PROFILES = List.of(V2RAYN, V2RAYNG, SHADOWROCKET); + + private final String userAgent; + private final String accept; + private final String acceptLanguage; + + SubscriptionClientProfile(String userAgent, String accept, String acceptLanguage) { + this.userAgent = userAgent; + this.accept = accept; + this.acceptLanguage = acceptLanguage; + } + + public String userAgent() { + return userAgent; + } + + public String accept() { + return accept; + } + + public String acceptLanguage() { + return acceptLanguage; + } + + /** + * 按账号与订阅格式确定性地选择一个身份。 + * + *

刻意不使用随机数：同一个账号每次都用同一个客户端，才符合真实用户的样子； + * 逐请求更换 User-Agent 本身就是更明显的机器特征。不同账号使用不同客户端， + * 又便于在必要时按账号区分上游流量。 + * + * @param accountId 子账号 ID，可为 null（视为 0） + * @param clash true 表示 Clash 格式，false 表示 V2Ray 格式 + */ + public static SubscriptionClientProfile forAccount(Integer accountId, boolean clash) { + List candidates = clash ? CLASH_PROFILES : V2_PROFILES; + int id = accountId == null ? 0 : accountId; + return candidates.get(Math.floorMod(id, candidates.size())); + } +} diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index 53bb996..f37b876 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -12,6 +12,12 @@ spring: datasource-cache: driver-class-name: org.sqlite.JDBC jdbc-url: jdbc:sqlite:cache.db + # 定时任务默认只有 1 个线程。订阅刷新一轮最多 2 个账号、每个 2 次请求（各 15 秒超时）， + # 最坏可占用约 1 分钟；单线程会让连接自检、cookie 检测等任务在此期间全部排队。 + task: + scheduling: + pool: + size: 2 mvc: view: prefix: /resources/templates/ @@ -38,6 +44,19 @@ local: subscription: cache-root: sub/accounts + # 分散刷新：每个账号在 window-hours 内至少刷新一次，两两之间至少相隔 min-gap-minutes； + # max-per-tick 限制单个 tick 刷新几个账号，避免重启后一批到期账号同时开火。 + refresh: + window-hours: 24 + min-gap-minutes: 60 + # 5 分钟一个轻量 tick，只处理已到期账号。 + tick-interval-ms: 300000 + # 重启后延迟 2 分钟再开始，先让应用完成启动。 + initial-delay-ms: 120000 + max-per-tick: 2 + # 失败后的重试间隔，必须小于 window-hours。 + retry-delay-minutes: 60 + stale-alert-multiplier: 2.0 upstream: # Use environment variables or an external config file in production. v2-url-template: "https://aaaa.gay/link/{key}?client=v2" diff --git a/src/test/java/com/lion/lionwebsite/Controller/PersonalControllerTest.java b/src/test/java/com/lion/lionwebsite/Controller/PersonalControllerTest.java index f8db837..b516e7e 100644 --- a/src/test/java/com/lion/lionwebsite/Controller/PersonalControllerTest.java +++ b/src/test/java/com/lion/lionwebsite/Controller/PersonalControllerTest.java @@ -107,19 +107,19 @@ class PersonalControllerTest { /** 手动更新订阅：成功与失败必须映射成不同的 result，前端据此提示。 */ @Test void updateSubReflectsServiceOutcome() throws Exception { - when(localService.updateSub(true)).thenReturn(true); + when(localService.updateSub()).thenReturn(true); var ok = mockMvc.perform(post("/personal/updateSub")) .andExpect(status().isOk()) .andReturn(); assertTrue(ok.getResponse().getContentAsString().contains("\"result\":\"success\"")); - when(localService.updateSub(true)).thenReturn(false); + when(localService.updateSub()).thenReturn(false); var failed = mockMvc.perform(post("/personal/updateSub")) .andExpect(status().isOk()) .andReturn(); assertTrue(failed.getResponse().getContentAsString().contains("\"result\":\"failure\"")); - verify(localService, times(2)).updateSub(true); + verify(localService, times(2)).updateSub(); } } diff --git a/src/test/java/com/lion/lionwebsite/Service/LocalServiceTest.java b/src/test/java/com/lion/lionwebsite/Service/LocalServiceTest.java index 7fc4f47..12807e9 100644 --- a/src/test/java/com/lion/lionwebsite/Service/LocalServiceTest.java +++ b/src/test/java/com/lion/lionwebsite/Service/LocalServiceTest.java @@ -30,7 +30,7 @@ class LocalServiceTest { private GalleryMapper galleryMapper; private PushService pushService; private RemoteService remoteService; - private SubscriptionRefreshService refreshService; + private SubscriptionRefreshScheduler refreshScheduler; private LocalService service; @BeforeEach @@ -40,9 +40,9 @@ class LocalServiceTest { galleryMapper = mock(GalleryMapper.class); pushService = mock(PushService.class); remoteService = mock(RemoteService.class); - refreshService = mock(SubscriptionRefreshService.class); + refreshScheduler = mock(SubscriptionRefreshScheduler.class); service = new LocalService(configurationMapper, shareFileMapper, galleryMapper, - pushService, remoteService, refreshService); + pushService, remoteService, refreshScheduler); } // ---------- CheckConnectionAvailability ---------- @@ -151,9 +151,9 @@ class LocalServiceTest { /** 全部子账号刷新成功：同步给节点并记录更新时间。 */ @Test void updateSubStampsTimeWhenAllAccountsSucceed() throws Exception { - when(refreshService.refreshAll()).thenReturn(true); + when(refreshScheduler.refreshAllNow()).thenReturn(true); - assertTrue(service.updateSub(true)); + assertTrue(service.updateSub()); verify(remoteService).requestSubscriptionSync(); verify(configurationMapper).updateConfiguration( @@ -163,23 +163,26 @@ class LocalServiceTest { /** 刷新失败时仍要通知节点，但不更新「上次更新时间」，避免掩盖故障。 */ @Test void updateSubDoesNotStampTimeWhenRefreshFails() throws Exception { - when(refreshService.refreshAll()).thenReturn(false); + when(refreshScheduler.refreshAllNow()).thenReturn(false); - assertFalse(service.updateSub(false)); + assertFalse(service.updateSub()); verify(remoteService).requestSubscriptionSync(); verify(configurationMapper, never()).updateConfiguration(anyString(), anyString()); } - /** 定时入口必须走同一套逻辑（手动与定时行为一致）。 */ + /** + * 定时刷新入口必须已经移除。 + * 旧的 {@code updateSubScheduler} 会一次性刷新全部账号且没有 initialDelay， + * 于是每次重启都触发一遍全量刷新——这正是本次要消除的行为。 + */ @Test - void scheduledUpdateDelegatesToSamePath() throws Exception { - when(refreshService.refreshAll()).thenReturn(true); - - service.updateSubScheduler(); - - verify(refreshService).refreshAll(); - verify(remoteService).requestSubscriptionSync(); + void scheduledFullRefreshEntryPointIsGone() { + assertThrows(NoSuchMethodException.class, + () -> LocalService.class.getMethod("updateSubScheduler")); + assertThrows(NoSuchMethodException.class, + () -> LocalService.class.getMethod("updateSub", boolean.class), + "不应再保留 isManual 参数：定时路径已不存在"); } // ---------- checkShareCode ---------- @@ -219,11 +222,8 @@ class LocalServiceTest { assertCron("verifyCookie", "0 0 0 * * *"); assertCron("checkShareCode", "0 0 4 * * *"); assertCron("clearThumbnailCache", "0 0 4 1 * *"); - - var m = LocalService.class.getMethod("updateSubScheduler"); - assertEquals(86400000L, - m.getAnnotation(org.springframework.scheduling.annotation.Scheduled.class).fixedRate(), - "订阅更新应为每 24 小时一次"); + // 订阅刷新已改为 SubscriptionRefreshScheduler 的分散 tick， + // 其周期与初始延迟由 SubscriptionRefreshSchedulerTest 锁定。 } private static void assertCron(String method, String expected) throws Exception { diff --git a/src/test/java/com/lion/lionwebsite/Service/SubServiceTest.java b/src/test/java/com/lion/lionwebsite/Service/SubServiceTest.java index ddf1ddd..f29eb80 100644 --- a/src/test/java/com/lion/lionwebsite/Service/SubServiceTest.java +++ b/src/test/java/com/lion/lionwebsite/Service/SubServiceTest.java @@ -37,7 +37,14 @@ class SubServiceTest { } private static SubscriptionAccount account(Integer id, String name, String key, boolean enabled) { - return new SubscriptionAccount(id, name, key, false, enabled, null, null, null, null, 0, null, null); + SubscriptionAccount a = new SubscriptionAccount(); + a.setId(id); + a.setName(name); + a.setUpstreamKey(key); + a.setFilterHighMultiplier(false); + a.setEnabled(enabled); + a.setBoundUserCount(0); + return a; } private static boolean ok(String json) { diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionFilteringTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionFilteringTest.java index 452e163..f507532 100644 --- a/src/test/java/com/lion/lionwebsite/Service/SubscriptionFilteringTest.java +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionFilteringTest.java @@ -63,8 +63,8 @@ class SubscriptionFilteringTest { } private void stubUpstream(List v2, List clash) throws Exception { - doReturn(v2).when(service).download(contains("client=v2")); - doReturn(clash).when(service).download(contains("client=clashmeta")); + doReturn(v2).when(service).download(contains("client=v2"), any()); + doReturn(clash).when(service).download(contains("client=clashmeta"), any()); } // ---------- URL 模板 ---------- @@ -108,7 +108,7 @@ class SubscriptionFilteringTest { String decodedV2 = new String(Base64.getMimeDecoder() .decode(Files.readString(v2)), StandardCharsets.UTF_8); assertTrue(decodedV2.contains("vmess://node-a"), "实际内容: " + decodedV2); - verify(subMapper).markSubscriptionRefreshSuccess(1); + verify(subMapper).markSubscriptionRefreshSuccess(eq(1), anyLong()); verify(subMapper, never()).markSubscriptionRefreshFailure(anyInt(), anyString()); // 临时文件不应残留 assertFalse(Files.exists(cacheRoot.resolve("1/clash.yaml.tmp"))); @@ -123,7 +123,7 @@ class SubscriptionFilteringTest { when(subMapper.selectSubscriptionAccount(2)).thenReturn(null); assertFalse(service.refresh(2)); - verify(subMapper, never()).markSubscriptionRefreshSuccess(anyInt()); + verify(subMapper, never()).markSubscriptionRefreshSuccess(anyInt(), anyLong()); } // ---------- 高倍率过滤（v2） ---------- @@ -264,12 +264,12 @@ class SubscriptionFilteringTest { void refreshFailsWhenUpstreamReturnsEmpty() throws Exception { SubscriptionAccount acc = account(12, "key12", true, false); when(subMapper.selectSubscriptionAccount(12)).thenReturn(acc); - doReturn(new ArrayList()).when(service).download(anyString()); + doReturn(new ArrayList()).when(service).download(anyString(), any()); assertFalse(service.refresh(12)); verify(subMapper).markSubscriptionRefreshFailure(eq(12), contains("为空")); - verify(subMapper, never()).markSubscriptionRefreshSuccess(anyInt()); + verify(subMapper, never()).markSubscriptionRefreshSuccess(anyInt(), anyLong()); assertFalse(Files.exists(cacheRoot.resolve("12/v2ray.txt")), "失败时不应产出缓存"); } @@ -291,7 +291,7 @@ class SubscriptionFilteringTest { void refreshFailsOnNetworkError() throws Exception { SubscriptionAccount acc = account(14, "key14", true, false); when(subMapper.selectSubscriptionAccount(14)).thenReturn(acc); - doThrow(new java.io.IOException("connection refused")).when(service).download(anyString()); + doThrow(new java.io.IOException("connection refused")).when(service).download(anyString(), any()); assertFalse(service.refresh(14)); @@ -303,7 +303,7 @@ class SubscriptionFilteringTest { void refreshTruncatesLongErrorMessages() throws Exception { SubscriptionAccount acc = account(15, "key15", true, false); when(subMapper.selectSubscriptionAccount(15)).thenReturn(acc); - doThrow(new java.io.IOException("x".repeat(900))).when(service).download(anyString()); + doThrow(new java.io.IOException("x".repeat(900))).when(service).download(anyString(), any()); assertFalse(service.refresh(15)); @@ -312,17 +312,20 @@ class SubscriptionFilteringTest { assertEquals(500, captor.getValue().length(), "错误信息应截断到 500 字符"); } - // ---------- refreshAll ---------- + // ---------- 全量刷新 ---------- + // + // 「遍历所有启用账号」的批量入口已从本服务移除（它正是会一次性打满上游的爆发写法）， + // 现在由 SubscriptionRefreshScheduler 按账号到期逐个刷新。 + // 调度行为与失败隔离见 SubscriptionRefreshSchedulerTest。 - /** 只刷新启用账号；任一失败即整体返回 false，但其余账号仍继续刷新。 */ + /** + * 单个账号刷新失败时，其余账号仍应正常写入缓存——失败隔离是分散刷新能替代 + * 批量刷新的前提：一个坏账号不该拖垮其他人。 + */ @Test - void refreshAllSkipsDisabledAndReportsAnyFailure() throws Exception { + void oneAccountFailureDoesNotPreventOthersFromCaching() throws Exception { SubscriptionAccount ok = account(20, "k20", true, false); - SubscriptionAccount off = account(21, "k21", false, false); SubscriptionAccount bad = account(22, "k22", true, false); - when(subMapper.selectAllSubscriptionAccounts()) - .thenReturn(new ArrayList<>(List.of(ok, off, bad))); - when(subMapper.selectSubscriptionAccount(20)).thenReturn(ok); when(subMapper.selectSubscriptionAccount(22)).thenReturn(bad); doAnswer(inv -> { @@ -332,36 +335,15 @@ class SubscriptionFilteringTest { if (url.contains("client=v2")) return List.of(Base64.getEncoder().encodeToString("vmess://a".getBytes(StandardCharsets.UTF_8))); return List.of("proxies:"); - }).when(service).download(anyString()); + }).when(service).download(anyString(), any()); - assertFalse(service.refreshAll(), "存在失败账号时整体应为 false"); + assertTrue(service.refresh(20)); + assertFalse(service.refresh(22)); - verify(subMapper).markSubscriptionRefreshSuccess(20); - verify(subMapper, never()).selectSubscriptionAccount(21); // 停用账号不刷新 - assertTrue(Files.exists(cacheRoot.resolve("20/v2ray.txt"))); - } - - /** 全部成功时返回 true。 */ - @Test - void refreshAllReturnsTrueWhenEveryAccountSucceeds() throws Exception { - SubscriptionAccount a1 = account(30, "k30", true, false); - SubscriptionAccount a2 = account(31, "k31", true, false); - when(subMapper.selectAllSubscriptionAccounts()).thenReturn(new ArrayList<>(List.of(a1, a2))); - when(subMapper.selectSubscriptionAccount(30)).thenReturn(a1); - when(subMapper.selectSubscriptionAccount(31)).thenReturn(a2); - stubUpstream(v2Upstream("vmess://a"), List.of("proxies:")); - - assertTrue(service.refreshAll()); - - verify(subMapper, times(2)).markSubscriptionRefreshSuccess(anyInt()); - } - - /** 没有任何账号时视为成功（无事可做）。 */ - @Test - void refreshAllSucceedsWithNoAccounts() { - when(subMapper.selectAllSubscriptionAccounts()).thenReturn(new ArrayList<>()); - - assertTrue(service.refreshAll()); + verify(subMapper).markSubscriptionRefreshSuccess(eq(20), anyLong()); + verify(subMapper).markSubscriptionRefreshFailure(eq(22), anyString()); + assertTrue(Files.exists(cacheRoot.resolve("20/v2ray.txt")), "成功账号应落盘"); + assertFalse(Files.exists(cacheRoot.resolve("22/v2ray.txt")), "失败账号不应落盘"); } // ---------- 缓存状态 ---------- diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshPlannerTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshPlannerTest.java new file mode 100644 index 0000000..88e940a --- /dev/null +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshPlannerTest.java @@ -0,0 +1,338 @@ +package com.lion.lionwebsite.Service; + +import org.junit.jupiter.api.Test; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.List; +import java.util.Random; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 订阅刷新排程：把「一天刷一次」变成「分散到一天里、但保证 24 小时内每个账号都刷到」。 + * + *

这里锁定三条最关键的契约： + *

    + *
  1. 首次排程真的把账号铺开，不是挤在一起（否则等于没改）；
    2. + *
  2. 下一次刷新一定在窗口内，这是「24 小时内必刷一次」的硬保证；
    3. + *
  3. 兜底裁剪能兜住计划缺失、长期失败等异常，不会让账号无限期不刷。
    4. + *
+ * 全部用固定时钟与固定随机种子，结果可复现。 + */ +class SubscriptionRefreshPlannerTest { + + private static final ZoneId ZONE = ZoneId.of("Asia/Shanghai"); + private static final Instant BASE = Instant.parse("2026-09-15T02:00:00Z"); // 北京时间 10:00 + + private static SubscriptionRefreshPlanner planner(long seed, Duration window, Duration minGap, Duration tick) { + return new SubscriptionRefreshPlanner(Clock.fixed(BASE, ZONE), new Random(seed), window, minGap, tick); + } + + private static SubscriptionRefreshPlanner defaultPlanner(long seed) { + return planner(seed, Duration.ofHours(24), Duration.ofHours(1), Duration.ofMinutes(5)); + } + + private static List ids(int count) { + List list = new ArrayList<>(); + for (int i = 1; i <= count; i++) + list.add(i); + return list; + } + + // ---------- 首次排程：分散 ---------- + + /** 每个账号都必须拿到计划，且都落在「现在」到「现在 + 窗口」之间。 */ + @Test + void initialSchedulePlacesEveryAccountInsideWindow() { + var planner = defaultPlanner(1); + List planned = planner.initialSchedule(ids(12)); + + assertEquals(12, planned.size()); + long now = BASE.toEpochMilli(); + for (long at : planned) { + assertTrue(at >= now, "计划不应早于当前时刻: " + at); + assertTrue(at <= now + Duration.ofHours(24).toMillis(), "计划不应超出窗口: " + at); + } + } + + /** + * 12 个账号必须被铺开到明显超过「挨个背靠背」的跨度。 + * 旧实现里它们全部落在 16 秒内，这里要求至少铺开窗口的一半。 + */ + @Test + void initialScheduleSpreadsAccountsAcrossWindow() { + var planner = defaultPlanner(7); + List planned = planner.initialSchedule(ids(12)); + + long min = planned.stream().mapToLong(Long::longValue).min().orElseThrow(); + long max = planned.stream().mapToLong(Long::longValue).max().orElseThrow(); + long spread = max - min; + + assertTrue(spread >= Duration.ofHours(12).toMillis(), + "12 个账号应铺开至少 12 小时，实际跨度 " + Duration.ofMillis(spread).toHours() + " 小时"); + } + + /** 排序后相邻账号之间应保持最小间隔，避免两个账号在同一分钟内打上游。 */ + @Test + void initialScheduleRespectsMinimumGap() { + var planner = defaultPlanner(3); + List planned = new ArrayList<>(planner.initialSchedule(ids(12))); + planned.sort(Long::compareTo); + + long minGap = Duration.ofHours(1).toMillis(); + for (int i = 1; i < planned.size(); i++) { + long gap = planned.get(i) - planned.get(i - 1); + assertTrue(gap >= minGap, + "第 " + i + " 个间隔只有 " + Duration.ofMillis(gap).toMinutes() + " 分钟，应不少于 60 分钟"); + } + } + + /** 不同种子应给出不同排程；若与种子无关，说明随机化没生效。 */ + @Test + void initialScheduleDependsOnRandomSource() { + var a = defaultPlanner(1).initialSchedule(ids(12)); + var b = defaultPlanner(99).initialSchedule(ids(12)); + assertNotEquals(a, b, "不同随机种子不应得到完全相同的排程"); + } + + /** 同一种子必须可复现：这是重启后行为一致的前提。 */ + @Test + void initialScheduleIsReproducibleForSameSeed() { + assertEquals(defaultPlanner(42).initialSchedule(ids(12)), + defaultPlanner(42).initialSchedule(ids(12))); + } + + @Test + void initialScheduleHandlesEmptyInput() { + assertTrue(defaultPlanner(1).initialSchedule(List.of()).isEmpty()); + } + + /** 单账号时也应拿到一个窗口内的计划，不能抛异常。 */ + @Test + void initialScheduleHandlesSingleAccount() { + List planned = defaultPlanner(5).initialSchedule(ids(1)); + assertEquals(1, planned.size()); + assertTrue(planned.getFirst() >= BASE.toEpochMilli()); + } + + // ---------- 容量校验 ---------- + + @Test + void capacityIsSufficientWhenGapsFit() { + // 12 个账号、间隔 1 小时 → 需要 11 小时，24 小时窗口放得下 + assertTrue(defaultPlanner(1).isCapacitySufficient(12)); + } + + @Test + void capacityIsInsufficientWhenGapsDoNotFit() { + // 50 个账号、间隔 1 小时 → 需要 49 小时，放不下 + assertFalse(defaultPlanner(1).isCapacitySufficient(50)); + assertTrue(defaultPlanner(1).capacityMessage(50).contains("放不下")); + } + + /** 容量不足也不能让排程崩溃：仍要给出窗口内的计划。 */ + @Test + void initialScheduleStillWorksWhenCapacityInsufficient() { + var planner = defaultPlanner(1); + List planned = planner.initialSchedule(ids(200)); + assertEquals(200, planned.size()); + long now = BASE.toEpochMilli(); + for (long at : planned) + assertTrue(at >= now && at <= now + Duration.ofHours(24).toMillis(), + "容量不足时仍须落在窗口内: " + at); + } + + // ---------- 24 小时保证 ---------- + + /** 成功后下一次刷新必须严格在 24 小时以内（留出一个 tick 的余量）。 */ + @Test + void nextAfterSuccessStaysWithinWindow() { + var planner = defaultPlanner(1); + long success = BASE.toEpochMilli(); + long next = planner.nextAfterSuccess(success); + + assertTrue(next > success, "下一次应晚于本次成功"); + assertTrue(next - success < Duration.ofHours(24).toMillis(), + "相邻两次成功应小于 24 小时，实际 " + Duration.ofMillis(next - success).toMinutes() + " 分钟"); + } + + /** 相位自我维持：连续推进多次都不应突破窗口。 */ + @Test + void successiveRefreshesNeverExceedWindow() { + var planner = defaultPlanner(1); + long at = BASE.toEpochMilli(); + for (int i = 0; i < 40; i++) { + long next = planner.nextAfterSuccess(at); + assertTrue(next - at < Duration.ofHours(24).toMillis(), "第 " + i + " 次推进超出窗口"); + at = next; + } + } + + // ---------- 失败重试与兜底 ---------- + + /** + * 失败后应延后重试，而不是立刻重打上游。 + * 间隔是固定的（数据库没有「连续失败次数」字段，做不出真正的指数退避）。 + */ + @Test + void nextAfterFailureDelaysRetry() { + var planner = defaultPlanner(1); + long now = BASE.toEpochMilli(); + + long retry = planner.nextAfterFailure(now, now); + + assertTrue(retry > now, "失败后不应立即重试"); + assertTrue(retry - now <= Duration.ofHours(1).toMillis(), + "重试不应拖太久，实际 " + Duration.ofMillis(retry - now).toMinutes() + " 分钟"); + } + + /** 重试间隔必须小于窗口，否则「窗口内重试」无从谈起。 */ + @Test + void constructorRejectsRetryDelayLargerThanWindow() { + assertThrows(IllegalArgumentException.class, + () -> new SubscriptionRefreshPlanner(Clock.fixed(BASE, ZONE), new Random(1), + Duration.ofHours(24), Duration.ofHours(1), Duration.ofMinutes(5), Duration.ofHours(48))); + assertThrows(IllegalArgumentException.class, + () -> new SubscriptionRefreshPlanner(Clock.fixed(BASE, ZONE), new Random(1), + Duration.ofHours(24), Duration.ofHours(1), Duration.ofMinutes(5), Duration.ZERO)); + } + + /** 退避不得把账号推过兜底线：仍要在窗口内重试。 */ + @Test + void nextAfterFailureNeverExceedsWindow() { + var planner = defaultPlanner(1); + long lastSuccess = BASE.toEpochMilli(); + long now = lastSuccess + Duration.ofHours(23).toMillis(); // 已经很接近兜底线 + + long next = planner.nextAfterFailure(now, lastSuccess); + + assertTrue(next <= lastSuccess + Duration.ofHours(24).toMillis(), + "重试时刻不得越过「上次成功 + 窗口」"); + } + + /** 已越过兜底线时应立刻到期，这是「必须刷新」的判定。 */ + @Test + void isOverdueWhenWindowElapsed() { + var planner = defaultPlanner(1); + long lastSuccess = BASE.toEpochMilli(); + + assertFalse(planner.isOverdue(lastSuccess, lastSuccess + Duration.ofHours(23).toMillis())); + assertTrue(planner.isOverdue(lastSuccess, lastSuccess + Duration.ofHours(24).toMillis())); + assertTrue(planner.isOverdue(lastSuccess, lastSuccess + Duration.ofHours(30).toMillis())); + } + + /** 从未成功过的账号不参与过期判定，避免刚建好的账号被立刻强刷。 */ + @Test + void isOverdueIsFalseWithoutKnownSuccess() { + assertFalse(defaultPlanner(1).isOverdue(null, BASE.toEpochMilli())); + } + + + + // ---------- 到期判定与计划损坏兜底 ---------- + + /** 计划在未来：未到期，不应刷新（否则分散效果会被自己破坏）。 */ + @Test + void isDueIsFalseBeforeScheduledTime() { + var planner = defaultPlanner(1); + long now = BASE.toEpochMilli(); + assertFalse(planner.isDue(now + Duration.ofHours(3).toMillis(), now)); + } + + /** 计划已到或已过：到期。 */ + @Test + void isDueIsTrueAtOrAfterScheduledTime() { + var planner = defaultPlanner(1); + long now = BASE.toEpochMilli(); + assertTrue(planner.isDue(now, now), "正好到点应可刷新"); + assertTrue(planner.isDue(now - Duration.ofMinutes(1).toMillis(), now), "已过点应可刷新"); + } + + /** + * 计划被写到超过一个窗口之后，按「计划损坏」处理并立即到期。 + * 这是 24 小时保证的兜底：即使字段被手工写坏，账号也不会长期不刷。 + */ + @Test + void isDueTreatsFarFutureScheduleAsCorrupt() { + var planner = defaultPlanner(1); + long now = BASE.toEpochMilli(); + assertTrue(planner.isDue(now + Duration.ofDays(10).toMillis(), now), + "超出窗口的计划不可能是本调度器产生的，应立即到期"); + } + + /** 尚未排程视为立即到期，避免新账号永远不刷。 */ + @Test + void isDueWithoutScheduleIsImmediate() { + assertTrue(defaultPlanner(1).isDue(null, BASE.toEpochMilli())); + } + + /** + * 失败重试排在将来时必须被尊重：否则越线的账号会在每个 tick 都重打上游。 + * 重试间隔是 30 分钟到 4 小时，远小于窗口，不构成「计划损坏」。 + */ + @Test + void retryScheduleIsRespectedNotTreatedAsCorrupt() { + var planner = defaultPlanner(1); + long lastSuccess = BASE.toEpochMilli() - Duration.ofHours(30).toMillis(); // 已越线 + long now = BASE.toEpochMilli(); + long retry = planner.nextAfterFailure(now, lastSuccess); + + assertTrue(retry > now, "越线后的重试仍应排在将来"); + assertFalse(planner.isDue(retry, now), "重试尚未到点，不应重复刷新"); + } + + // ---------- 冷启动排程 ---------- + + /** 从未成功过的账号应尽快刷，而不是被分散到 20 小时后。 */ + @Test + void coldStartScheduleIsImmediateAndStaggered() { + var planner = defaultPlanner(1); + List planned = planner.coldStartSchedule(ids(5), Duration.ofMinutes(5)); + + long now = BASE.toEpochMilli(); + assertEquals(now, planned.getFirst(), "第一个应立即可刷"); + for (int i = 1; i < planned.size(); i++) { + assertEquals(Duration.ofMinutes(5).toMillis(), planned.get(i) - planned.get(i - 1), + "相邻账号应按 tick 间隔错开"); + } + assertTrue(planned.getLast() <= now + Duration.ofMinutes(30).toMillis(), + "全部应在半小时内排完，避免新账号长时间没有可用订阅"); + } + + @Test + void coldStartScheduleHandlesEmptyInput() { + assertTrue(defaultPlanner(1).coldStartSchedule(List.of(), Duration.ofMinutes(5)).isEmpty()); + } + + // ---------- 构造参数校验 ---------- + + @Test + void constructorRejectsNonPositiveWindow() { + assertThrows(IllegalArgumentException.class, + () -> planner(1, Duration.ZERO, Duration.ofHours(1), Duration.ofMinutes(5))); + assertThrows(IllegalArgumentException.class, + () -> planner(1, Duration.ofHours(-1), Duration.ofHours(1), Duration.ofMinutes(5))); + } + + @Test + void constructorRejectsNegativeGapOrTick() { + assertThrows(IllegalArgumentException.class, + () -> planner(1, Duration.ofHours(24), Duration.ofHours(-1), Duration.ofMinutes(5))); + assertThrows(IllegalArgumentException.class, + () -> planner(1, Duration.ofHours(24), Duration.ofHours(1), Duration.ofMinutes(-5))); + } + + /** 配置访问器应如实回传构造时的值，供调度器日志与告警阈值使用。 */ + @Test + void exposesConfiguredWindow() { + var planner = defaultPlanner(1); + assertEquals(Duration.ofHours(24), planner.window()); + assertEquals(Duration.ofHours(1), planner.minGap()); + assertEquals(Duration.ofMinutes(5), planner.tickInterval()); + assertEquals(BASE, planner.clock().instant()); + } +} diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshSchedulerTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshSchedulerTest.java new file mode 100644 index 0000000..0ff1b83 --- /dev/null +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshSchedulerTest.java @@ -0,0 +1,481 @@ +package com.lion.lionwebsite.Service; + +import com.lion.lionwebsite.Dao.normal.SubMapper; +import com.lion.lionwebsite.Domain.SubscriptionAccount; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.List; +import java.util.Random; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +/** + * 分散刷新的调度行为。 + * + *

这里锁定用户明确要求的几条：每个账号 24 小时内必刷一次、时间互相错开、 + * 以及「中途重启导致当天没刷」必须能自愈。全部使用固定时钟，不依赖真实时间流逝。 + */ +class SubscriptionRefreshSchedulerTest { + + private static final ZoneId ZONE = ZoneId.of("Asia/Shanghai"); + private static final Instant BASE = Instant.parse("2026-09-15T02:00:00Z"); + + private static final long HOUR = 3_600_000L; + private static final long MINUTE = 60_000L; + + private SubMapper subMapper; + private SubscriptionRefreshService refreshService; + private PushService pushService; + + @BeforeEach + void setUp() { + subMapper = mock(SubMapper.class); + refreshService = mock(SubscriptionRefreshService.class); + pushService = mock(PushService.class); + } + + /** 用固定时钟构造，使「现在」可控。 */ + private SubscriptionRefreshScheduler schedulerAt(Instant instant) { + SubscriptionRefreshPlanner planner = new SubscriptionRefreshPlanner( + Clock.fixed(instant, ZONE), new Random(42), + Duration.ofHours(24), Duration.ofHours(1), Duration.ofMinutes(5)); + SubscriptionRefreshScheduler scheduler = new SubscriptionRefreshScheduler( + subMapper, refreshService, planner, pushService); + scheduler.maxPerTick = 2; + scheduler.staleAlertMultiplier = 2.0; + return scheduler; + } + + private SubscriptionRefreshScheduler scheduler() { + return schedulerAt(BASE); + } + + private static SubscriptionAccount account(int id, Long nextRefreshAt, Long lastSuccessEpoch) { + SubscriptionAccount a = new SubscriptionAccount(); + a.setId(id); + a.setName("acc-" + id); + a.setUpstreamKey("key-" + id); + a.setEnabled(true); + a.setBoundUserCount(0); + a.setNextRefreshAt(nextRefreshAt); + a.setLastSuccessEpoch(lastSuccessEpoch); + return a; + } + + private void existing(SubscriptionAccount... accounts) { + ArrayList list = new ArrayList<>(List.of(accounts)); + when(subMapper.selectAllSubscriptionAccounts()).thenReturn(list); + } + + /** 刷新成功时返回 true。 */ + private void refreshSucceedsForAll() { + when(refreshService.refresh(anyInt())).thenReturn(true); + } + + // ---------- 到期判定 ---------- + + /** 计划时刻未到：本轮不应刷新，避免把「分散」又变回「每次都刷」。 */ + @Test + void accountNotDueIsSkipped() { + long future = BASE.toEpochMilli() + 5 * HOUR; + existing(account(1, future, BASE.toEpochMilli())); + scheduler().refreshDueAccounts(); + + verify(refreshService, never()).refresh(anyInt()); + } + + /** 计划时刻已到：必须刷新，并重排下一次。 */ + @Test + void accountDueIsRefreshedAndRescheduled() { + long past = BASE.toEpochMilli() - MINUTE; + existing(account(1, past, BASE.toEpochMilli() - 2 * HOUR)); + refreshSucceedsForAll(); + + int refreshed = scheduler().refreshDueAccounts(); + + assertEquals(1, refreshed); + verify(refreshService).refresh(1); + verify(subMapper).updateNextRefreshAt(eq(1), anyLong()); + } + + /** 停用的账号即使到期也不刷新。 */ + @Test + void disabledAccountsAreNeverRefreshed() { + SubscriptionAccount disabled = account(1, BASE.toEpochMilli() - HOUR, null); + disabled.setEnabled(false); + existing(disabled); + + assertEquals(0, scheduler().refreshDueAccounts()); + verify(refreshService, never()).refresh(anyInt()); + } + + // ---------- 24 小时硬保证 ---------- + + /** + * 兜底线：即使计划缺失（例如手工清了字段），只要距上次成功已过一个窗口， + * 也必须立即刷新。这是「24 小时内必刷一次」的最后一道闸。 + */ + @Test + void overdueAccountIsRefreshedEvenWithoutSchedule() { + long lastSuccess = BASE.toEpochMilli() - 25 * HOUR; + existing(account(1, null, lastSuccess)); + refreshSucceedsForAll(); + + assertEquals(1, scheduler().refreshDueAccounts()); + verify(refreshService).refresh(1); + } + + /** 到期但未越线的账号若计划被写得很晚，兜底裁剪要能把它拉回来。 */ + @Test + void overdueAccountBeatsFarFutureSchedule() { + long lastSuccess = BASE.toEpochMilli() - 25 * HOUR; + long farFuture = BASE.toEpochMilli() + 30 * 24 * HOUR; + existing(account(1, farFuture, lastSuccess)); + refreshSucceedsForAll(); + + assertEquals(1, scheduler().refreshDueAccounts(), "越线账号必须无视过晚的计划立即刷新"); + verify(refreshService).refresh(1); + } + + /** 成功后重排的下一次必须在 24 小时以内，锁定「必刷一次」的硬上限。 */ + @Test + void rescheduledTimeStaysWithinWindow() { + long past = BASE.toEpochMilli() - MINUTE; + existing(account(1, past, BASE.toEpochMilli() - 2 * HOUR)); + refreshSucceedsForAll(); + + scheduler().refreshDueAccounts(); + + ArgumentCaptor next = ArgumentCaptor.forClass(Long.class); + verify(subMapper).updateNextRefreshAt(eq(1), next.capture()); + long delta = next.getValue() - BASE.toEpochMilli(); + assertTrue(delta > 0, "下一次应在未来"); + assertTrue(delta < 24 * HOUR, "下一次必须在 24 小时内，实际 " + delta / HOUR + " 小时"); + } + + // ---------- 重启兜底 ---------- + + /** + * 重启后的追赶：停机 6 小时期间错过的账号，恢复后应被补刷。 + * 这是用户明确要求的「中途重启导致当天没更新」兜底。 + */ + @Test + void accountsMissedDuringDowntimeAreCaughtUpAfterRestart() { + long downtimeStart = BASE.toEpochMilli() - 6 * HOUR; + // 三个账号的计划都落在停机窗口里，即「本来该刷但进程没在跑」 + existing( + account(1, downtimeStart, BASE.toEpochMilli() - 20 * HOUR), + account(2, downtimeStart + HOUR, BASE.toEpochMilli() - 20 * HOUR), + account(3, downtimeStart + 2 * HOUR, BASE.toEpochMilli() - 20 * HOUR)); + refreshSucceedsForAll(); + + // 一个 tick 最多 2 个：先补 2 个，避免同时开火 + assertEquals(2, scheduler().refreshDueAccounts()); + } + + /** + * 连续多个 tick 应能把停机期间积压的账号全部排空（不会永久卡住）。 + * + *

mock 的 mapper 不会像数据库那样保存状态，因此这里显式模拟持久化： + * {@code updateNextRefreshAt} 会把新时刻写回对象，成功时同时更新 lastSuccessEpoch。 + * 否则被刷过的账号在内存里仍显示为「到期」，测出的排空行为没有意义。 + */ + @Test + void catchUpDrainsBacklogAcrossTicks() { + long downtimeStart = BASE.toEpochMilli() - 6 * HOUR; + List accounts = new ArrayList<>(List.of( + account(1, downtimeStart, BASE.toEpochMilli() - 20 * HOUR), + account(2, downtimeStart + HOUR, BASE.toEpochMilli() - 20 * HOUR), + account(3, downtimeStart + 2 * HOUR, BASE.toEpochMilli() - 20 * HOUR), + account(4, downtimeStart + 3 * HOUR, BASE.toEpochMilli() - 20 * HOUR), + account(5, downtimeStart + 4 * HOUR, BASE.toEpochMilli() - 20 * HOUR))); + when(subMapper.selectAllSubscriptionAccounts()).thenReturn(new ArrayList<>(accounts)); + refreshSucceedsForAll(); + persistSchedules(accounts); + + var scheduler = scheduler(); + assertEquals(2, scheduler.refreshDueAccounts(), "第一轮最多 2 个"); + assertEquals(2, scheduler.refreshDueAccounts(), "第二轮再 2 个"); + assertEquals(1, scheduler.refreshDueAccounts(), "第三轮补齐剩余 1 个"); + verify(refreshService, times(5)).refresh(anyInt()); + } + + /** + * 模拟数据库的持久化：把 updateNextRefreshAt 写回对象。 + * 成功刷新还必须推进 lastSuccessEpoch，否则兜底线会一直把账号判为到期。 + */ + private void persistSchedules(List accounts) { + doAnswer(invocation -> { + int id = invocation.getArgument(0); + long next = invocation.getArgument(1); + long now = BASE.toEpochMilli(); + for (SubscriptionAccount a : accounts) { + if (a.getId() == id) { + a.setNextRefreshAt(next); + a.setLastSuccessEpoch(now); + a.setLastError(null); + } + } + return null; + }).when(subMapper).updateNextRefreshAt(anyInt(), anyLong()); + } + + // ---------- 单轮上限 ---------- + + /** 单轮刷新数量必须有上限，否则重启后一批到期账号会同时打上游。 */ + @Test + void maxPerTickLimitsConcurrentRefreshes() { + List accounts = new ArrayList<>(); + for (int i = 1; i <= 12; i++) + accounts.add(account(i, BASE.toEpochMilli() - HOUR, BASE.toEpochMilli() - 30 * HOUR)); + when(subMapper.selectAllSubscriptionAccounts()).thenReturn(new ArrayList<>(accounts)); + refreshSucceedsForAll(); + + assertEquals(2, scheduler().refreshDueAccounts()); + verify(refreshService, times(2)).refresh(anyInt()); + } + + // ---------- 失败处理 ---------- + + /** 单个账号失败不影响其他账号继续刷新。 */ + @Test + void oneFailureDoesNotBlockOthers() { + existing( + account(1, BASE.toEpochMilli() - HOUR, BASE.toEpochMilli() - 30 * HOUR), + account(2, BASE.toEpochMilli() - HOUR, BASE.toEpochMilli() - 30 * HOUR)); + when(refreshService.refresh(1)).thenReturn(false); + when(refreshService.refresh(2)).thenReturn(true); + + assertEquals(2, scheduler().refreshDueAccounts(), "失败的账号也要计入本轮尝试"); + + verify(refreshService).refresh(1); + verify(refreshService).refresh(2); + // 失败账号也要重排，否则会每个 tick 反复重打上游 + verify(subMapper).updateNextRefreshAt(eq(1), anyLong()); + } + + /** 失败后的重试时刻应晚于当前，避免失败的账号被连续重打。 */ + @Test + void failureReschedulesIntoFuture() { + existing(account(1, BASE.toEpochMilli() - HOUR, BASE.toEpochMilli() - 30 * HOUR)); + when(refreshService.refresh(1)).thenReturn(false); + + scheduler().refreshDueAccounts(); + + ArgumentCaptor next = ArgumentCaptor.forClass(Long.class); + verify(subMapper).updateNextRefreshAt(eq(1), next.capture()); + assertTrue(next.getValue() > BASE.toEpochMilli(), "重试时刻应在未来"); + } + + /** 手动全量刷新不受单轮上限限制：管理页按钮的语义就是立刻全部刷新。 */ + @Test + void refreshAllNowIgnoresPerTickLimit() { + List accounts = new ArrayList<>(); + for (int i = 1; i <= 5; i++) + accounts.add(account(i, BASE.toEpochMilli() + 10 * HOUR, BASE.toEpochMilli())); + when(subMapper.selectAllSubscriptionAccounts()).thenReturn(new ArrayList<>(accounts)); + refreshSucceedsForAll(); + + assertTrue(scheduler().refreshAllNow()); + verify(refreshService, times(5)).refresh(anyInt()); + } + + /** + * 手动刷新后必须重排计划。否则「刚手动刷完」会让当天的计划作废， + * 该账号反而可能超过 24 小时没有下次更新。 + */ + @Test + void refreshAllNowReschedulesEveryAccount() { + existing(account(1, BASE.toEpochMilli() + 10 * HOUR, BASE.toEpochMilli()), + account(2, BASE.toEpochMilli() + 10 * HOUR, BASE.toEpochMilli())); + refreshSucceedsForAll(); + + scheduler().refreshAllNow(); + + verify(subMapper).updateNextRefreshAt(eq(1), anyLong()); + verify(subMapper).updateNextRefreshAt(eq(2), anyLong()); + } + + /** 手动刷新中只要有一个失败，整体结果即为失败（页面据此提示）。 */ + @Test + void refreshAllNowReportsFailureIfAnyAccountFails() { + existing(account(1, BASE.toEpochMilli(), BASE.toEpochMilli()), + account(2, BASE.toEpochMilli(), BASE.toEpochMilli())); + when(refreshService.refresh(1)).thenReturn(true); + when(refreshService.refresh(2)).thenReturn(false); + + assertFalse(scheduler().refreshAllNow()); + } + + /** 停用账号不参与手动全量刷新。 */ + @Test + void refreshAllNowSkipsDisabled() { + SubscriptionAccount disabled = account(1, BASE.toEpochMilli(), BASE.toEpochMilli()); + disabled.setEnabled(false); + existing(disabled); + + assertTrue(scheduler().refreshAllNow()); + verify(refreshService, never()).refresh(anyInt()); + } + + // ---------- 计划补齐 ---------- + + /** + * 从未成功过的账号（没有计划）应被自动排上并尽快刷新，而不是永远不刷。 + * 会被写入两次计划：一次是冷启动排程，一次是刷新成功后的重排。 + */ + @Test + void missingScheduleIsFilledInAndRefreshedPromptly() { + existing(account(1, null, null)); + refreshSucceedsForAll(); + + scheduler().refreshDueAccounts(); + + verify(refreshService).refresh(1); + ArgumentCaptor next = ArgumentCaptor.forClass(Long.class); + verify(subMapper, times(2)).updateNextRefreshAt(eq(1), next.capture()); + assertTrue(next.getAllValues().getFirst() <= BASE.toEpochMilli() + 5 * MINUTE, + "从未成功过的账号应尽快刷新"); + } + + /** 补齐的计划应落在窗口内，且不会让账号立刻被刷（避免与建号时的首刷重复）。 */ + @Test + void filledScheduleLandsInsideWindow() { + existing(account(1, null, null)); + + // 只跑到补齐逻辑：不提供成功桩，refreshDueAccounts 仍会先补齐计划 + when(refreshService.refresh(1)).thenReturn(true); + scheduler().refreshDueAccounts(); + + ArgumentCaptor next = ArgumentCaptor.forClass(Long.class); + verify(subMapper, atLeastOnce()).updateNextRefreshAt(eq(1), next.capture()); + long earliest = next.getAllValues().stream().mapToLong(Long::longValue).min().orElseThrow(); + assertTrue(earliest <= BASE.toEpochMilli() + 24 * HOUR, "补齐的计划必须落在窗口内"); + } + + /** + * 回归测试：刚被铺开计划的账号不应在同一轮里就被刷新。 + * + *

曾经的缺陷是「把计划写进数据库却忘了更新内存里的对象」——这些账号在本轮 + * 仍带着 null 计划，被判定为立即到期，于是迁移后首批账号会在第一个 tick + * 全部集中刷新，铺开形同虚设。 + */ + @Test + void newlyScheduledAccountsAreNotRefreshedInSameTick() { + List accounts = new ArrayList<>(); + for (int i = 1; i <= 5; i++) + accounts.add(account(i, null, BASE.toEpochMilli() - 2 * HOUR)); // 曾成功过 → 按窗口铺开 + when(subMapper.selectAllSubscriptionAccounts()).thenReturn(new ArrayList<>(accounts)); + refreshSucceedsForAll(); + + int refreshed = scheduler().refreshDueAccounts(); + + assertEquals(0, refreshed, "刚铺开计划的账号本轮不应立即刷新"); + verify(refreshService, never()).refresh(anyInt()); + assertEquals(5, accounts.stream().filter(a -> a.getNextRefreshAt() != null).count(), + "铺开的计划应写回内存对象"); + } + + /** 已有计划的账号不应被重新排程，否则每次 tick 都会打乱分散效果。 */ + @Test + void existingScheduleIsNotOverwritten() { + long future = BASE.toEpochMilli() + 8 * HOUR; + existing(account(1, future, BASE.toEpochMilli())); + + scheduler().refreshDueAccounts(); + + verify(subMapper, never()).updateNextRefreshAt(anyInt(), anyLong()); + } + + // ---------- 陈旧告警 ---------- + + /** 超过两个窗口仍未成功应告警，让运维在用户投诉前发现问题。 */ + @Test + void staleAccountTriggersAlert() { + long lastSuccess = BASE.toEpochMilli() - 50 * HOUR; + SubscriptionAccount stale = account(1, BASE.toEpochMilli() + HOUR, lastSuccess); + stale.setLastError("上游 HTTP 状态码 403"); + existing(stale); + + scheduler().refreshDueAccounts(); + + verify(pushService).sendToMe(contains("未成功刷新")); + } + + /** 告警只发一次，避免每个 tick 都刷屏。 */ + @Test + void staleAlertIsSentOnlyOnce() { + long lastSuccess = BASE.toEpochMilli() - 50 * HOUR; + existing(account(1, BASE.toEpochMilli() + HOUR, lastSuccess)); + + var scheduler = scheduler(); + scheduler.refreshDueAccounts(); + scheduler.refreshDueAccounts(); + scheduler.refreshDueAccounts(); + + verify(pushService, times(1)).sendToMe(anyString()); + } + + /** 未陈旧的账号不应告警。 */ + @Test + void healthyAccountDoesNotAlert() { + existing(account(1, BASE.toEpochMilli() + HOUR, BASE.toEpochMilli() - HOUR)); + + scheduler().refreshDueAccounts(); + + verify(pushService, never()).sendToMe(anyString()); + } + + /** 账号恢复正常后，若再次陈旧应能重新告警。 */ + @Test + void alertResetsAfterSuccessfulRefresh() { + long lastSuccess = BASE.toEpochMilli() - 50 * HOUR; + SubscriptionAccount stale = account(1, BASE.toEpochMilli() - MINUTE, lastSuccess); + when(subMapper.selectAllSubscriptionAccounts()) + .thenReturn(new ArrayList<>(List.of(stale))); + when(refreshService.refresh(1)).thenReturn(true, false); + + var scheduler = scheduler(); + scheduler.refreshDueAccounts(); // 告警 + 成功 → 重置 + scheduler.refreshDueAccounts(); // 失败，但尚未再次陈旧 + + verify(pushService, times(1)).sendToMe(anyString()); + } + + // ---------- 调度注解契约 ---------- + + /** + * tick 必须用 fixedDelay（一轮跑完再计下一轮）而不是 fixedRate， + * 并且必须有 initialDelay——这正是旧实现「每次重启立刻全量重刷」的根因。 + */ + @Test + void tickUsesFixedDelayWithInitialDelay() throws Exception { + var annotation = SubscriptionRefreshScheduler.class.getMethod("tick") + .getAnnotation(org.springframework.scheduling.annotation.Scheduled.class); + + assertNotNull(annotation, "tick 应带 @Scheduled"); + assertEquals("${subscription.refresh.tick-interval-ms:300000}", annotation.fixedDelayString()); + assertEquals("${subscription.refresh.initial-delay-ms:120000}", annotation.initialDelayString()); + assertEquals(-1, annotation.fixedRate(), "不应再使用 fixedRate（未设置时为 -1）"); + } + + /** + * 旧的「每 24 小时一次全量」入口必须已经消失。 + * 它没有 initialDelay，会在每次重启时立刻重刷全部账号。 + */ + @Test + void localServiceNoLongerSchedulesFullRefresh() { + assertThrows(NoSuchMethodException.class, + () -> LocalService.class.getMethod("updateSubScheduler"), + "LocalService 不应再暴露 24 小时全量刷新入口"); + } +} diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshServiceTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshServiceTest.java index 34388ed..9cc53d5 100644 --- a/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshServiceTest.java +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshServiceTest.java @@ -11,19 +11,27 @@ import static org.junit.jupiter.api.Assertions.*; import static org.mockito.Mockito.*; class SubscriptionRefreshServiceTest { + + private static SubscriptionAccount account(String key) { + SubscriptionAccount a = new SubscriptionAccount(); + a.setId(1); + a.setName("sample"); + a.setUpstreamKey(key); + a.setEnabled(true); + a.setBoundUserCount(0); + return a; + } @Test void stalledDownloadDoesNotHoldStateLockAndStaleResultIsDiscarded(@TempDir Path directory) throws Exception { SubMapper mapper = mock(SubMapper.class); - SubscriptionAccount original = new SubscriptionAccount(1, "sample", "old", false, true, - null, null, null, null, 0, null, null); - SubscriptionAccount changed = new SubscriptionAccount(1, "sample", "new", false, true, - null, null, null, null, 0, null, null); + SubscriptionAccount original = account("old"); + SubscriptionAccount changed = account("new"); when(mapper.selectSubscriptionAccount(1)).thenReturn(original); var coordinator = new SubscriptionStateCoordinator(); CountDownLatch started = new CountDownLatch(1); CountDownLatch release = new CountDownLatch(1); var service = new SubscriptionRefreshService(mapper, coordinator) { - @Override List download(String url) throws java.io.IOException { + @Override List download(String url, com.lion.lionwebsite.Util.SubscriptionClientProfile profile) throws java.io.IOException { started.countDown(); try { if (!release.await(5, TimeUnit.SECONDS)) throw new java.io.IOException("test timed out"); @@ -47,7 +55,7 @@ class SubscriptionRefreshServiceTest { release.countDown(); assertFalse(refresh.get(2, TimeUnit.SECONDS)); assertFalse(service.hasCompleteCache(1)); - verify(mapper, never()).markSubscriptionRefreshSuccess(any()); + verify(mapper, never()).markSubscriptionRefreshSuccess(any(), anyLong()); } finally { release.countDown(); worker.shutdownNow(); } } } diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshWiringTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshWiringTest.java new file mode 100644 index 0000000..98df964 --- /dev/null +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshWiringTest.java @@ -0,0 +1,97 @@ +package com.lion.lionwebsite.Service; + +import com.lion.lionwebsite.Configuration.CustomBean; +import com.lion.lionwebsite.Dao.normal.SubMapper; +import org.junit.jupiter.api.Test; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.mock; + +/** + * 新调度组件的 Spring 装配。 + * + *

纯 Mockito 测试不会发现「bean 无法装配」这类问题——线上表现是整个应用起不来， + * 因此这里用真实的 Spring 容器把 {@link SubscriptionRefreshScheduler} 装一遍。 + * 只注册必要的 bean，不加载 Web 与数据源，避免测试依赖数据库与端口。 + */ +class SubscriptionRefreshWiringTest { + + @Configuration + static class Stub { + @Bean + SubMapper subMapper() { + return mock(SubMapper.class); + } + + @Bean + SubscriptionStateCoordinator subscriptionStateCoordinator() { + return new SubscriptionStateCoordinator(); + } + + @Bean + SubscriptionRefreshService subscriptionRefreshService(SubMapper subMapper, + SubscriptionStateCoordinator coordinator) { + return new SubscriptionRefreshService(subMapper, coordinator); + } + + @Bean + PushService pushService() { + return mock(PushService.class); + } + + @Bean + SubscriptionRefreshPlanner subscriptionRefreshPlanner() { + return new SubscriptionRefreshPlanner(java.time.Clock.systemDefaultZone(), new java.util.Random(), + Duration.ofHours(24), Duration.ofHours(1), Duration.ofMinutes(5)); + } + + @Bean + SubscriptionRefreshScheduler subscriptionRefreshScheduler(SubMapper subMapper, + SubscriptionRefreshService refreshService, + SubscriptionRefreshPlanner planner, + PushService pushService) { + return new SubscriptionRefreshScheduler(subMapper, refreshService, planner, pushService); + } + } + + /** 调度器必须能从容器里装配出来，且依赖都已就位。 */ + @Test + void schedulerBeanIsWired() { + try (var ctx = new AnnotationConfigApplicationContext(Stub.class)) { + SubscriptionRefreshScheduler scheduler = ctx.getBean(SubscriptionRefreshScheduler.class); + assertNotNull(scheduler); + assertNotNull(scheduler.planner()); + assertEquals(Duration.ofHours(24), scheduler.planner().window()); + } + } + + /** + * 生产用的 planner bean 必须真正读到 application.yaml 里的值。 + * 这里直接调用 {@link CustomBean} 的工厂方法验证默认值与注入方向一致， + * 避免「配置写了但没人读」以及「读错键」两类静默故障。 + */ + @Test + void plannerFactoryReadsConfiguredWindow() { + SubscriptionRefreshPlanner planner = new CustomBean() + .subscriptionRefreshPlanner(24, 60, 300_000, 60); + + assertEquals(Duration.ofHours(24), planner.window()); + assertEquals(Duration.ofMinutes(60), planner.minGap()); + assertEquals(Duration.ofMinutes(5), planner.tickInterval()); + } + + /** 12 个账号的现状下，默认配置（24 小时窗口、60 分钟间隔）应当放得下。 */ + @Test + void defaultConfigurationFitsCurrentAccountCount() { + SubscriptionRefreshPlanner planner = new CustomBean() + .subscriptionRefreshPlanner(24, 60, 300_000, 60); + + assertTrue(planner.isCapacitySufficient(12), + "当前 12 个子账号在默认配置下应能铺开"); + } +} diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotServiceTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotServiceTest.java index 0b90d65..bed4489 100644 --- a/src/test/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotServiceTest.java +++ b/src/test/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotServiceTest.java @@ -23,8 +23,13 @@ import static org.junit.jupiter.api.Assertions.*; class SubscriptionStandbySnapshotServiceTest { @Test void buildsDeterministicSnapshotWithoutPlainPublicKeys(@TempDir Path directory) throws Exception { - SubscriptionAccount account = new SubscriptionAccount(1, "account", "upstream-secret", true, true, - null, null, null, null, 2, null, null); + SubscriptionAccount account = new SubscriptionAccount(); + account.setId(1); + account.setName("account"); + account.setUpstreamKey("upstream-secret"); + account.setFilterHighMultiplier(true); + account.setEnabled(true); + account.setBoundUserCount(2); ArrayList accounts = new ArrayList<>(java.util.List.of(account)); ArrayList bindings = new ArrayList<>(java.util.List.of( new SubBind("public-key-a", "user-a", 1, "account", true, true), diff --git a/src/test/java/com/lion/lionwebsite/Util/SubscriptionClientProfileTest.java b/src/test/java/com/lion/lionwebsite/Util/SubscriptionClientProfileTest.java new file mode 100644 index 0000000..9b57a2d --- /dev/null +++ b/src/test/java/com/lion/lionwebsite/Util/SubscriptionClientProfileTest.java @@ -0,0 +1,94 @@ +package com.lion.lionwebsite.Util; + +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 抓取上游订阅时伪装的客户端身份。 + * + *

改造前上游看到的固定是 {@code Apache-HttpClient/5.6.4 (Java/25.0.4)}， + * 相当于自报「我是服务器上的 Java 程序」。这里锁定三条： + * 不再出现 Java/HttpClient 字样、同一账号身份稳定、两种格式用各自合理的客户端。 + */ +class SubscriptionClientProfileTest { + + /** 任何伪装都必须先去掉 HttpClient 与 Java 版本的自报家门。 */ + @Test + void noProfileLeaksJavaOrHttpClientIdentity() { + for (SubscriptionClientProfile profile : SubscriptionClientProfile.values()) { + String ua = profile.userAgent(); + assertFalse(ua.contains("Java"), "不能出现 Java 版本: " + ua); + assertFalse(ua.contains("HttpClient"), "不能出现 HttpClient: " + ua); + assertFalse(ua.contains("Apache"), "不能出现 Apache: " + ua); + assertFalse(ua.isBlank(), "User-Agent 不能为空"); + } + } + + /** 身份必须确定：同一账号每次都是同一个客户端，才像真实用户。 */ + @Test + void profileIsStableForSameAccount() { + for (int id = 1; id <= 20; id++) { + assertEquals(SubscriptionClientProfile.forAccount(id, false), + SubscriptionClientProfile.forAccount(id, false)); + assertEquals(SubscriptionClientProfile.forAccount(id, true), + SubscriptionClientProfile.forAccount(id, true)); + } + } + + /** 参数为 null 时不应抛异常（新建账号尚未拿到 ID 的场景）。 */ + @Test + void nullAccountIdIsTolerated() { + assertNotNull(SubscriptionClientProfile.forAccount(null, false)); + assertNotNull(SubscriptionClientProfile.forAccount(null, true)); + } + + /** 不同账号应分散到不同客户端，便于必要时按账号区分上游流量。 */ + @Test + void differentAccountsShareTheProfilePoolEvenly() { + List clash = List.of(SubscriptionClientProfile.values()).stream() + .filter(p -> p.userAgent().toLowerCase().contains("clash") || p.userAgent().toLowerCase().contains("mihomo")) + .toList(); + assertFalse(clash.isEmpty(), "应存在 Clash 系客户端身份"); + + var first = SubscriptionClientProfile.forAccount(1, true); + var second = SubscriptionClientProfile.forAccount(2, true); + assertNotEquals(first, second, "相邻账号应使用不同客户端身份"); + } + + /** Clash 格式不应伪装成 V2Ray 客户端，反之亦然。 */ + @Test + void clashAndV2ProfilesAreDistinct() { + for (int id = 1; id <= 10; id++) { + SubscriptionClientProfile clash = SubscriptionClientProfile.forAccount(id, true); + SubscriptionClientProfile v2 = SubscriptionClientProfile.forAccount(id, false); + assertNotEquals(clash, v2, "同一账号的两种格式应使用不同客户端身份"); + + String clashUa = clash.userAgent().toLowerCase(); + assertTrue(clashUa.contains("clash") || clashUa.contains("mihomo"), + "Clash 订阅应伪装成 Clash 系客户端: " + clash.userAgent()); + + String v2Ua = v2.userAgent().toLowerCase(); + assertTrue(v2Ua.contains("v2ray") || v2Ua.contains("shadowrocket"), + "V2Ray 订阅应伪装成 V2Ray 系客户端: " + v2.userAgent()); + } + } + + /** 负数 ID 也不能越界（防御性：ID 由数据库提供，但不应因异常值崩溃）。 */ + @Test + void negativeAccountIdStaysInRange() { + assertNotNull(SubscriptionClientProfile.forAccount(-5, false)); + assertNotNull(SubscriptionClientProfile.forAccount(-5, true)); + } + + /** 每个身份都应带上 Accept 与 Accept-Language，避免头部组合明显异常。 */ + @Test + void everyProfileCarriesPlausibleAcceptHeaders() { + for (SubscriptionClientProfile profile : SubscriptionClientProfile.values()) { + assertFalse(profile.accept().isBlank(), "Accept 不能为空"); + assertTrue(profile.acceptLanguage().contains("zh"), "应声明中文偏好: " + profile.acceptLanguage()); + } + } +}