修复订阅快照校验、回退与原生反射配置

llm_readme.txt

@@ -94,6 +94,7 @@ storageNode/
 | 6 | `IdentityMessage` | 握手 |
 | 7 | `MaintainMessage` | 心跳 |
 | 8 | `AvailableCheckMessage` | 可用性检查 |
+| 9 | `SubscriptionSnapshotMessage` | 主站 → 节点，完整订阅备机快照 |
 
 ---
 
@@ -107,10 +108,12 @@ storageNode/
 
 ### BackupSubServer（端口 8889）
 - 提供 V2Ray 和 Clash 代理订阅文件
-- 每 12 小时从配置的 URL 下载最新订阅数据
+- 不直接访问上游；主站完成下载和倍率过滤后，通过 Netty 类型 9 推送完整快照
-- 过滤高比例代理节点
+- 按公开 Key 的 SHA-256 查找用户绑定的子账号，未知 Key 返回 404
-- 文件存储于 `sub/DouNaiV2ray.txt` 和 `sub/DouNaiClash.txt`
+- 快照经 SHA-256 和 HMAC 校验后原子落盘，主站离线时继续分发最后成功版本
-- 通过路径密钥进行身份验证
+- 文件存储于 `sub/snapshots/{revision}/accounts/{accountId}/`
+- 没有有效快照或快照超过最大有效期时返回 503，不回退旧共享订阅
+- `GET /health/subscription` 提供不含 Key 和订阅正文的快照状态
 
 ---
 
@@ -132,10 +135,18 @@ storageNode/
 
 ### config.properties（从 `/root/gallery/storageNode/config.properties` 加载）
 ```properties
-DouNaiV2ray=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=v2
+SubscriptionSyncEnabled=false
-DouNaiClash=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=clashmeta
+SubscriptionSyncSecret=
+SubscriptionDataDir=/root/gallery/storageNode/sub
+SubscriptionMaxStaleSeconds=604800
+SubscriptionMaxPayloadBytes=52428800
+SubscriptionHttpPort=8889
+SubscriptionHttpWorkers=4
+SubscriptionSocketTimeoutMs=10000
 ```
 
+生产同步密钥优先通过 `SUBSCRIPTION_SYNC_SECRET` 环境变量提供，不得提交到仓库或写入日志。
+
 ### simplelogger.properties
 - 日志级别：`info`
 - 时间戳格式：`yyyy-MM-dd HH:mm:ss`
@@ -150,4 +161,4 @@ DouNaiClash=https://aaaa.gay/link/X7zEqkIx5gtIGugO?client=clashmeta
 - 外部连接：`lionwebsite.xyz`、`personal.lionwebsite.xyz`、`aaaa.gay`
 - GraalVM 原生镜像编译，包含 Jackson 反射配置
 - 大量使用 Lombok（`@Data`、`@Slf4j`）
-- 使用 Hutool 工具库处理文件/ZIP/HTTP 操作
+- 使用 Hutool 工具库处理文件/ZIP/HTTP 操作

src/main/java/lion/Externel/BackupSubServer.java

@@ -11,8 +11,7 @@ import java.net.Socket;
 import java.net.URLDecoder;
 import java.nio.charset.StandardCharsets;
 import java.util.*;
-import java.util.concurrent.ExecutorService;
+import java.util.concurrent.*;
-import java.util.concurrent.Executors;
 
 /** HTTP distributor for the last-known-good subscription snapshot. */
 @Slf4j
@@ -24,17 +23,24 @@ public final class BackupSubServer implements Runnable {
     public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount) {
         this.snapshotStore = Objects.requireNonNull(snapshotStore);
         this.port = port;
-        this.workers = Executors.newFixedThreadPool(Math.max(1, workerCount));
+        int workersCount = Math.max(1, workerCount);
+        this.workers = new ThreadPoolExecutor(workersCount, workersCount, 0, TimeUnit.MILLISECONDS,
+                new ArrayBlockingQueue<>(workersCount * 32), new ThreadPoolExecutor.AbortPolicy());
     }
 
     @Override
     public void run() {
-        snapshotStore.load();
         try (ServerSocket serverSocket = new ServerSocket(port)) {
             log.info("备机订阅服务监听端口 {}", port);
             while (!Thread.currentThread().isInterrupted()) {
                 Socket socket = serverSocket.accept();
-                workers.execute(() -> handle(socket));
+                try {
+                    workers.execute(() -> handle(socket));
+                } catch (RejectedExecutionException e) {
+                    try (socket) {
+                        send(socket, 503, "Service Unavailable", "text/plain", new byte[0], false);
+                    } catch (IOException ignored) { }
+                }
             }
         } catch (IOException e) {
             log.error("备机订阅服务停止: {}", e.getMessage());

src/main/java/lion/Main.java

@@ -15,6 +15,7 @@ public class Main {
         SubscriptionSnapshotStore snapshotStore = new SubscriptionSnapshotStore(
                 java.nio.file.Paths.get(Config.subscriptionDataDir), Config.subscriptionSyncSecret,
                 Config.subscriptionMaxStaleSeconds, Config.subscriptionMaxPayloadBytes);
+        snapshotStore.load();
         new Thread(new BackupSubServer(snapshotStore, Config.subscriptionHttpPort, Config.subscriptionHttpWorkers),
                 "subscription-backup-http").start();
         new Thread(() -> MultiThreadedHTTPServer.main(null)).start();

src/main/java/lion/Message/Main/SubscriptionSnapshotMessage.java

@@ -2,10 +2,12 @@ package lion.Message.Main;
 
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import lombok.ToString;
 import lion.Message.AbstractMessage;
 
 @Data
 @NoArgsConstructor
+@ToString(exclude = {"payloadBase64", "signature"})
 public class SubscriptionSnapshotMessage extends AbstractMessage {
     {
         messageType = SUBSCRIPTION_SNAPSHOT_MESSAGE;

src/main/java/lion/Service/SubscriptionSnapshotStore.java

@@ -46,35 +46,47 @@ public final class SubscriptionSnapshotStore {
     }
 
     public void load() {
+        List<String> candidates = new ArrayList<>();
         try {
             Path pointer = root.resolve("current-revision");
-            if (!Files.isRegularFile(pointer)) {
+            if (Files.isRegularFile(pointer)) {
-                log.warn("没有找到订阅快照，备机订阅暂不可用");
+                String revision = Files.readString(pointer, StandardCharsets.UTF_8).trim();
-                return;
+                if (isRevision(revision)) candidates.add(revision);
+            }
+            Path snapshots = root.resolve("snapshots");
+            if (Files.isDirectory(snapshots)) {
+                try (var stream = Files.list(snapshots)) {
+                    stream.filter(Files::isDirectory)
+                            .map(path -> path.getFileName().toString())
+                            .filter(SubscriptionSnapshotStore::isRevision)
+                            .filter(revision -> !candidates.contains(revision))
+                            .sorted(Comparator.reverseOrder())
+                            .forEach(candidates::add);
+                }
             }
-            String revision = Files.readString(pointer, StandardCharsets.UTF_8).trim();
-            if (!isRevision(revision))
-                throw new IOException("当前快照 revision 非法");
-            Snapshot snapshot = loadSnapshot(root.resolve("snapshots").resolve(revision));
-            current.set(snapshot);
-            log.info("加载订阅快照成功 revision={} accounts={} bindings={}", shortRevision(revision), snapshot.accountCount(), snapshot.bindingCount());
         } catch (Exception e) {
-            current.set(null);
+            log.warn("扫描订阅快照失败: {}", e.getMessage());
-            log.error("加载订阅快照失败，备机订阅暂不可用: {}", e.getMessage());
         }
+        for (String revision : candidates) {
+            try {
+                Snapshot snapshot = loadSnapshot(root.resolve("snapshots").resolve(revision));
+                current.set(snapshot);
+                writePointer(revision);
+                log.info("加载订阅快照成功 revision={} accounts={} bindings={}", shortRevision(revision), snapshot.accountCount(), snapshot.bindingCount());
+                return;
+            } catch (Exception e) {
+                log.warn("订阅快照损坏，尝试上一版本 revision={}: {}", shortRevision(revision), e.getMessage());
+            }
+        }
+        current.set(null);
+        log.warn("没有可用的订阅快照，备机订阅暂不可用");
     }
 
     public ApplyResult apply(SubscriptionSnapshotMessage message) {
+        Path staging = null;
         try {
             if (message == null || message.getSchemaVersion() != 1 || !isRevision(message.getRevision()))
                 return new ApplyResult(APPLY_INVALID, "消息版本或 revision 非法");
-            Snapshot old = current.get();
-            if (old != null) {
-                if (message.getRevision().equals(old.revision()))
-                    return new ApplyResult(APPLY_OLD, "revision 已存在");
-                if (message.getGeneratedAt() < old.generatedAt())
-                    return new ApplyResult(APPLY_OLD, "快照时间早于当前版本");
-            }
             if (syncSecret.length == 0)
                 return new ApplyResult(APPLY_INVALID, "同步密钥未配置");
 
@@ -91,10 +103,17 @@ public final class SubscriptionSnapshotStore {
                 return new ApplyResult(APPLY_INVALID, "revision 与 payload 不一致");
             SubscriptionSnapshotPayload payload = objectMapper.readValue(payloadBytes, SubscriptionSnapshotPayload.class);
             SnapshotData data = validatePayload(payload);
+            Snapshot old = current.get();
+            if (old != null) {
+                if (message.getRevision().equals(old.revision()))
+                    return new ApplyResult(APPLY_OLD, "revision 已存在");
+                if (message.getGeneratedAt() < old.generatedAt())
+                    return new ApplyResult(APPLY_OLD, "快照时间早于当前版本");
+            }
 
             Path snapshots = root.resolve("snapshots");
             Files.createDirectories(snapshots);
-            Path staging = snapshots.resolve(".staging-" + message.getRevision());
+            staging = snapshots.resolve(".staging-" + message.getRevision());
             deleteRecursively(staging);
             Files.createDirectories(staging.resolve("accounts"));
             for (AccountFiles account : data.accounts.values()) {
@@ -106,17 +125,15 @@ public final class SubscriptionSnapshotStore {
             Map<String, Object> manifest = new LinkedHashMap<>();
             manifest.put("revision", message.getRevision());
             manifest.put("generatedAt", message.getGeneratedAt());
-            manifest.put("payload", payload);
+            // 保留主站签名对应的原始 JSON 字节，重启时不依赖 Jackson 再序列化顺序。
+            manifest.put("payloadBase64", Base64.getEncoder().encodeToString(payloadBytes));
             Files.write(staging.resolve("manifest.json"), objectMapper.writeValueAsBytes(manifest), StandardOpenOption.CREATE_NEW);
 
             Path destination = snapshots.resolve(message.getRevision());
             if (Files.exists(destination))
                 deleteRecursively(destination);
             atomicMove(staging, destination);
-            Path pointerTmp = root.resolve("current-revision.tmp");
+            writePointer(message.getRevision());
-            Files.writeString(pointerTmp, message.getRevision() + "\n", StandardCharsets.UTF_8,
-                    StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
-            atomicMove(pointerTmp, root.resolve("current-revision"));
 
             Snapshot snapshot = new Snapshot(message.getRevision(), message.getGeneratedAt(), data.byKeyHash,
                     data.accounts.size(), data.bindingCount);
@@ -126,9 +143,22 @@ public final class SubscriptionSnapshotStore {
         } catch (Exception e) {
             log.error("应用订阅快照失败: {}", e.getMessage());
             return new ApplyResult(APPLY_IO_ERROR, e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage());
+        } finally {
+            if (staging != null && Files.exists(staging)) {
+                try { deleteRecursively(staging); }
+                catch (IOException e) { log.warn("清理订阅快照 staging 失败: {}", staging); }
+            }
         }
     }
 
+    private void writePointer(String revision) throws IOException {
+        Files.createDirectories(root);
+        Path pointerTmp = root.resolve("current-revision.tmp");
+        Files.writeString(pointerTmp, revision + "\n", StandardCharsets.UTF_8,
+                StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING);
+        atomicMove(pointerTmp, root.resolve("current-revision"));
+    }
+
     public Lookup lookup(String client, String publicKey) {
         Snapshot snapshot = current.get();
         if (snapshot == null || snapshot.expired(System.currentTimeMillis(), maxStaleMillis))
@@ -152,9 +182,10 @@ public final class SubscriptionSnapshotStore {
         JsonNode manifest = objectMapper.readTree(Files.readAllBytes(directory.resolve("manifest.json")));
         String revision = manifest.path("revision").asText();
         long generatedAt = manifest.path("generatedAt").asLong(0);
-        SubscriptionSnapshotPayload payload = objectMapper.treeToValue(manifest.path("payload"), SubscriptionSnapshotPayload.class);
+        byte[] payloadBytes = decodeBase64(manifest.path("payloadBase64").asText(null), maxPayloadBytes);
+        SubscriptionSnapshotPayload payload = objectMapper.readValue(payloadBytes, SubscriptionSnapshotPayload.class);
         SnapshotData data = validatePayload(payload);
-        if (!isRevision(revision) || !revision.equals(sha256(objectMapper.writeValueAsBytes(payload))))
+        if (!isRevision(revision) || !constantEquals(revision, sha256(payloadBytes)))
             throw new IOException("快照 manifest revision 校验失败");
         Map<Integer, AccountFiles> accounts = new HashMap<>();
         for (SubscriptionAccountSnapshot account : payload.getAccounts()) {
@@ -203,7 +234,10 @@ public final class SubscriptionSnapshotStore {
             Path snapshots = root.resolve("snapshots");
             List<Path> dirs;
             try (var stream = Files.list(snapshots)) {
-                dirs = stream.filter(Files::isDirectory).filter(p -> !p.getFileName().toString().startsWith(".staging-")).toList();
+                dirs = stream.filter(Files::isDirectory)
+                        .filter(p -> !p.getFileName().toString().startsWith(".staging-"))
+                        .sorted(Comparator.comparingLong(SubscriptionSnapshotStore::lastModified).reversed())
+                        .toList();
             }
             dirs.stream().filter(p -> !p.getFileName().toString().equals(currentRevision))
                     .skip(1).forEach(p -> {
@@ -264,6 +298,11 @@ public final class SubscriptionSnapshotStore {
     private static boolean isBase64Sha(String value) { return value != null && value.matches("[0-9a-fA-F]{64}"); }
     private static String shortRevision(String revision) { return revision == null ? null : revision.substring(0, Math.min(12, revision.length())); }
 
+    private static long lastModified(Path path) {
+        try { return Files.getLastModifiedTime(path).toMillis(); }
+        catch (IOException e) { return 0; }
+    }
+
     private static void atomicMove(Path source, Path target) throws IOException {
         try { Files.move(source, target, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING); }
         catch (AtomicMoveNotSupportedException e) { Files.move(source, target, StandardCopyOption.REPLACE_EXISTING); }

src/main/java/lion/storageNode.java

@@ -1,5 +1,6 @@
 package lion;
 
+import lion.Config.Config;
 import lion.Domain.GalleryTask;
 import lion.Message.*;
 import lion.Message.Main.*;
@@ -150,7 +151,10 @@ public class storageNode {
 
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object msg) {
-            log.info(String.valueOf(msg));
+            if (msg instanceof SubscriptionSnapshotMessage snapshot)
+                log.info("收到订阅快照 revision={}", shortRevision(snapshot.getRevision()));
+            else
+                log.info(String.valueOf(msg));
             AbstractMessage abstractMessage = (AbstractMessage) msg;
 
             switch (abstractMessage.messageType){
@@ -201,7 +205,7 @@ public class storageNode {
                 }
                 case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> {
                     SubscriptionSnapshotMessage snapshotMessage = (SubscriptionSnapshotMessage) abstractMessage;
-                    if (!ctx.channel().equals(server)) {
+                    if (!Config.subscriptionSyncEnabled || !ctx.channel().equals(server)) {
                         ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, SubscriptionSnapshotStore.APPLY_INVALID));
                         return;
                     }

src/main/resources/reflect-config.json

@@ -62,6 +62,51 @@
     "allDeclaredFields" : true,
     "allPublicFields" : true
   },
+  {
+    "name": "lion.Message.Main.SubscriptionSnapshotMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.SubscriptionSnapshotPayload",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.SubscriptionAccountSnapshot",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Message.Main.SubscriptionBindingSnapshot",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
+  {
+    "name": "lion.Service.SubscriptionSnapshotStore$Status",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "allDeclaredFields" : true,
+    "allPublicFields" : true
+  },
   {
     "name": "lion.Message.AbstractMessage",
     "allDeclaredConstructors" : true,
@@ -80,4 +125,4 @@
     "allDeclaredFields" : true,
     "allPublicFields" : true
   }
-]
+]

src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java

@@ -28,7 +28,7 @@ class SubscriptionSnapshotStoreTest {
     @Test
     void appliesSnapshotAndServesByHashedPublicKey(@TempDir Path directory) throws Exception {
         SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
-        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", 1000);
+        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis());
 
         assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code());
         SubscriptionSnapshotStore.Lookup v2 = store.lookup("v2", "public-key-1");
@@ -41,7 +41,7 @@ class SubscriptionSnapshotStoreTest {
     @Test
     void rejectsTamperedPayloadAndKeepsPreviousSnapshot(@TempDir Path directory) throws Exception {
         SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
-        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", 1000);
+        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis());
         assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code());
         message.setPayloadBase64(Base64.getEncoder().encodeToString("tampered".getBytes(StandardCharsets.UTF_8)));
         assertEquals(SubscriptionSnapshotStore.APPLY_INVALID, store.apply(message).code());
@@ -50,7 +50,7 @@ class SubscriptionSnapshotStoreTest {
 
     @Test
     void loadsLastGoodSnapshotAfterRestart(@TempDir Path directory) throws Exception {
-        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", 1000);
+        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis());
         SubscriptionSnapshotStore first = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
         assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, first.apply(message).code());
         SubscriptionSnapshotStore second = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);