固定 Java 21 release 编译并更新项目说明

安全修复：
- netty-all 4.1.101.Final -> 4.1.138.Final：4.1.101 存在多个已知 CVE
  （netty-codec-http 修复线 4.1.108/4.1.125/4.1.129/4.1.132-4.1.137，
  netty-common 修复线 4.1.115/4.1.118）。该组件直接面向主站与公网。
- commons-compress 1.25.0 -> 1.28.0（CVE-2024-26308/25710 修复于 1.26.0）。
- logback-classic 1.4.14 -> 1.5.38，slf4j 2.0.9 -> 2.0.19。
- jackson-databind 2.15.2 -> 2.22.2：留在 Jackson 2.x 线而非迁 3，因为本组件
  负责解析并落盘快照数据，且与主站（已升 Jackson 3）的跨版本线上格式已实测兼容。
- hutool 5.8.26 -> 5.8.47；lombok 1.18.40 -> 1.18.48；junit 5.10.2 -> 5.14.4。
- native-maven-plugin 0.10.3 -> 1.1.8（配合 GraalVM 25）。

移除 org.apache.httpcomponents:httpclient 4.5.14：全仓库无任何引用
（HTTP 调用走 hutool 的 HttpRequest），且该线自 2022-11 起已 EOL。

验证：mvn test 5 项全过；隔离冒烟加载线上真实快照目录（53 个文件）——
Jackson 2.22.2 正确解析由 2.15.2 写出的 manifest，/health/subscription 报
state=ready、12 账号/12 绑定，/sub/v2 与 /sub/cat 分别 43024/125045 字节
（与升级前完全一致）；与 Boot 4 主站（Jackson 3）端到端联调，
订阅快照 result=0（APPLY_SUCCESS）。

.gitignore

@@ -36,3 +36,7 @@ build/
 
 ### Mac OS ###
 .DS_Store
+/.idea/encodings.xml
+
+# 本地运行与测试日志
+run.out

llm_readme.txt

@@ -0,0 +1,165 @@
+# storageNode 项目结构
+
+## 概述
+分布式存储/下载节点服务端，用于画廊网站。职责包括：从远程源下载图片集、压缩为 ZIP 归档、通过 HTTP 提供下载（支持断点续传）、管理代理订阅配置（V2Ray/Clash），并通过 Netty 自定义 TCP 协议与中心服务器通信。
+
+- **Group ID:** `org.lion`
+- **Version:** `1.0`
+- **Java 版本:** 21 字节码（生产使用 GraalVM JDK 25）
+- **构建工具:** Maven（单模块）
+- **编译目标:** 当前以 JVM/JAR + `lib/` 运行；保留 GraalVM 原生配置但尚未完成 JDK 25 原生验证
+
+---
+
+## 目录结构
+
+```
+storageNode/
+├── pom.xml                                              # Maven 构建配置
+└── src/
+    ├── main/
+    │   ├── java/
+    │   │   └── lion/
+    │   │       ├── Main.java                            # 程序入口
+    │   │       ├── storageNode.java                     # Netty TCP 核心节点
+    │   │       ├── CustomUtil.java                      # 工具方法（HTTP 通知、端口查找、ObjectMapper）
+    │   │       ├── MultiThreadedHTTPServer.java          # HTTP 文件下载服务（端口 8888）
+    │   │       ├── Config/
+    │   │       │   └── Config.java                      # 加载 config.properties
+    │   │       ├── Domain/
+    │   │       │   └── GalleryTask.java                  # 下载任务领域模型
+    │   │       ├── ErrorCode/
+    │   │       │   └── ErrorCode.java                   # 错误码常量
+    │   │       ├── Externel/
+    │   │       │   └── BackupSubServer.java             # 代理订阅文件服务（端口 8889）
+    │   │       ├── Message/
+    │   │       │   ├── AbstractMessage.java             # 消息基类 + 消息类型常量
+    │   │       │   ├── MessageCodec.java                # Netty 编解码器（ByteBuf ↔ AbstractMessage，JSON 格式）
+    │   │       │   └── Main/
+    │   │       │       ├── AvailableCheckMessage.java   # 可用性检查（type=8）
+    │   │       │       ├── DeleteGalleryMessage.java    # 删除画廊请求（type=3）
+    │   │       │       ├── DownloadPostMessage.java     # 提交下载任务（type=1）
+    │   │       │       ├── DownloadStatusMessage.java   # 上报任务状态（type=2）
+    │   │       │       ├── IdentityMessage.java         # 身份认证握手（type=6）
+    │   │       │       ├── MaintainMessage.java         # 心跳维持（type=7）
+    │   │       │       └── ResponseMessage.java         # 通用响应（type=0）
+    │   │       └── Service/
+    │   │           ├── DeleteService.java               # 删除画廊目录
+    │   │           └── DownloadCheckService.java        # 下载监控与压缩服务
+    │   └── resources/
+    │       ├── config.properties                        # DouNai 订阅地址配置
+    │       ├── simplelogger.properties                  # SLF4J 日志配置（输出到 run.out）
+    │       └── reflect-config.json                      # GraalVM 反射配置（Jackson 序列化）
+    └── test/
+        └── java/                                        # 订阅快照与下载/压缩恢复测试
+```
+
+---
+
+## 关键依赖
+
+| 依赖 | 版本 | 用途 |
+|---|---|---|
+| `io.netty:netty-all` | 4.1.138.Final | TCP 服务端/客户端 |
+| `com.fasterxml.jackson.core:jackson-databind` | 2.22.2 | JSON 序列化 |
+| `org.projectlombok:lombok` | 1.18.48 | 简化样板代码（`@Data`, `@Slf4j`） |
+| `ch.qos.logback:logback-classic` | 1.5.38 | 日志实现 |
+| `cn.hutool:hutool-all` | 5.8.47 | 文件操作、ZIP 压缩、HTTP 请求 |
+| `org.apache.commons:commons-compress` | 1.28.0 | 压缩归档 |
+| `org.graalvm.buildtools:native-maven-plugin` | 1.1.8 | 原生镜像配置（待验证） |
+
+---
+
+## 启动流程
+
+1. `Main.main()` → 调用 `boot()`（遗留的 Netty Bootstrap），然后执行 `Config.loadConfig()`
+2. 在新线程中启动 `BackupSubServer`（端口 8889）— 提供代理订阅文件下载
+3. 在新线程中启动 `MultiThreadedHTTPServer`（端口 8888）— 提供画廊 ZIP 文件下载
+4. 主线程创建 `storageNode()` 实例（阻塞构造函数，永不返回）：
+   - 从端口 26321 开始查找空闲端口，绑定 Netty TCP 服务端
+   - 作为 TCP 客户端连接 `lionwebsite.xyz:26322~26342`，唤醒中心服务器
+   - 启动 `DownloadCheckService` 和 5 秒定时任务 `mainThread()`
+
+---
+
+## 通信协议（Netty TCP）
+
+自定义协议格式：`[1字节类型] + [4字节长度] + [JSON 负载]`
+
+| 类型字节 | 消息类 | 方向 |
+|---|---|---|
+| 0 | `ResponseMessage` | 响应 |
+| 1 | `DownloadPostMessage` | 服务端 → 节点 |
+| 2 | `DownloadStatusMessage` | 节点 → 服务端 |
+| 3 | `DeleteGalleryMessage` | 服务端 → 节点 |
+| 6 | `IdentityMessage` | 握手 |
+| 7 | `MaintainMessage` | 心跳 |
+| 8 | `AvailableCheckMessage` | 可用性检查 |
+| 9 | `SubscriptionSnapshotMessage` | 主站 → 节点，完整订阅备机快照 |
+
+---
+
+## HTTP 服务
+
+### MultiThreadedHTTPServer（端口 8888）
+- 提供压缩后的画廊 ZIP 文件下载
+- 仅接受来自 `lionwebsite.xyz` IP 的连接
+- 支持 HTTP 206 Partial Content（断点续传）
+- 参数：`AuthCode`（管理员访问）、`gid`（画廊 ID）
+
+### BackupSubServer（端口 8889）
+- 提供 V2Ray 和 Clash 代理订阅文件
+- 不直接访问上游；主站完成下载和倍率过滤后，通过 Netty 类型 9 推送完整快照
+- 按公开 Key 的 SHA-256 查找用户绑定的子账号，未知 Key 返回 404
+- 快照经 SHA-256 和 HMAC 校验后原子落盘，主站离线时继续分发最后成功版本
+- 文件存储于 `sub/snapshots/{revision}/accounts/{accountId}/`
+- 没有有效快照或快照超过最大有效期时返回 503，不回退旧共享订阅
+- `GET /health/subscription` 提供不含 Key 和订阅正文的快照状态
+
+---
+
+## 核心服务
+
+### DownloadCheckService
+- 扫描下载目录（`/root/gallery/hentai/download/`）监控进度
+- 通过检测 `galleryinfo.txt` 文件判断下载完成
+- 将完成的下载任务移入压缩队列
+- 后台线程每 5 秒执行 ZIP 压缩，完成后删除源目录
+
+### DeleteService
+- 按名称删除画廊目录
+- 失败时返回 `ErrorCode.IO_ERROR` 或 `ErrorCode.FILE_NOT_FOUND`
+
+---
+
+## 配置说明
+
+### config.properties（从 `/root/gallery/storageNode/config.properties` 加载）
+```properties
+SubscriptionSyncEnabled=false
+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`
+- 输出文件：`run.out`
+
+---
+
+## 架构说明
+
+- 单模块 Maven 项目，现有 5 个测试用例覆盖订阅快照和压缩失败恢复
+- 硬编码文件系统路径（`/root/gallery/...`）→ 仅限 Linux 部署
+- 外部连接：`lionwebsite.xyz`、`personal.lionwebsite.xyz`、`aaaa.gay`
+- GraalVM 原生镜像编译，包含 Jackson 反射配置
+- 大量使用 Lombok（`@Data`、`@Slf4j`）
+- 使用 Hutool 工具库处理文件/ZIP/HTTP 操作

pom.xml

@@ -9,8 +9,7 @@
     <version>1.0</version>
 
     <properties>
-        <maven.compiler.source>21</maven.compiler.source>
+        <maven.compiler.release>21</maven.compiler.release>
-        <maven.compiler.target>21</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 
@@ -18,50 +17,76 @@
         <dependency>
             <groupId>io.netty</groupId>
             <artifactId>netty-all</artifactId>
-            <version>4.1.101.Final</version>
+            <version>4.1.138.Final</version>
         </dependency>
 
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
-            <version>2.15.2</version>
+            <version>2.22.2</version>
         </dependency>
 
         <dependency>
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
-            <version>1.18.30</version>
+            <version>1.18.48</version>
         </dependency>
 
         <dependency>
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
-            <version>2.0.9</version>
+            <version>2.0.19</version>
         </dependency>
 
 
         <dependency>
-            <groupId>org.slf4j</groupId>
+            <groupId>ch.qos.logback</groupId>
-            <artifactId>slf4j-simple</artifactId>
+            <artifactId>logback-classic</artifactId>
-            <version>2.0.7</version>
+            <version>1.5.38</version>
         </dependency>
 
 
         <dependency>
             <groupId>cn.hutool</groupId>
             <artifactId>hutool-all</artifactId>
-            <version>5.8.26</version>
+            <version>5.8.47</version>
         </dependency>
 
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-compress</artifactId>
-            <version>1.25.0</version>
+            <version>1.28.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <version>5.14.4</version>
+            <scope>test</scope>
         </dependency>
     </dependencies>
 
     <build>
         <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>3.2.5</version>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.13.0</version>
+                <configuration>
+                    <annotationProcessorPaths>
+                        <path>
+                            <groupId>org.projectlombok</groupId>
+                            <artifactId>lombok</artifactId>
+                            <version>1.18.48</version>
+                        </path>
+                    </annotationProcessorPaths>
+                </configuration>
+            </plugin>
 <!--            <plugin>-->
 <!--                <groupId>org.apache.maven.plugins</groupId>-->
 <!--                <artifactId>maven-assembly-plugin</artifactId>-->
@@ -88,21 +113,21 @@
             <plugin>
                 <groupId>org.graalvm.buildtools</groupId>
                 <artifactId>native-maven-plugin</artifactId>
-                <version>0.9.28</version>
+                <version>1.1.8</version>
                 <configuration>
                     <mainClass>lion.Main</mainClass>
                     <imageName>storageNode</imageName>
                     <buildArgs>
                         <arg>-H:+ReportExceptionStackTraces</arg>
-                        <arg>--gc=G1</arg>
                         <arg>--enable-url-protocols=https</arg>
-                        <arg>-H:IncludeResources="simplelogger.properties"</arg>
+                        <arg>-H:IncludeResources="logback.xml"</arg>
-                        <arg>--initialize-at-build-time=org.slf4j.simple.SimpleLogger,org.slf4j.simple.SimpleLoggerFactory</arg>
+                        <arg>--initialize-at-build-time=ch.qos.logback.classic,ch.qos.logback.core,ch.qos.logback.classic.pattern,ch.qos.logback.core.pattern</arg>
                         <arg>-H:ReflectionConfigurationFiles=src/main/resources/reflect-config.json</arg>
                     </buildArgs>
                     <metadataRepository>
                         <enabled>true</enabled>
                     </metadataRepository>
+
                 </configuration>
             </plugin>
         </plugins>

src/main/java/lion/Config/Config.java

@@ -0,0 +1,45 @@
+package lion.Config;
+
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class Config {
+    public static boolean subscriptionSyncEnabled;
+    public static String subscriptionSyncSecret;
+    public static String subscriptionDataDir;
+    public static long subscriptionMaxStaleSeconds;
+    public static int subscriptionMaxPayloadBytes;
+    public static int subscriptionHttpPort;
+    public static int subscriptionHttpWorkers;
+    public static int subscriptionSocketTimeoutMs;
+
+    public static void loadConfig(){
+        Properties prop = new Properties();
+
+        try (InputStream input = new FileInputStream("/root/gallery/storageNode/config.properties")) {
+            prop.load(input);
+            subscriptionSyncEnabled = Boolean.parseBoolean(value(prop, "SubscriptionSyncEnabled", "false"));
+            subscriptionSyncSecret = System.getenv().getOrDefault("SUBSCRIPTION_SYNC_SECRET",
+                    value(prop, "SubscriptionSyncSecret", ""));
+            subscriptionDataDir = value(prop, "SubscriptionDataDir", "/root/gallery/storageNode/sub");
+            subscriptionMaxStaleSeconds = Long.parseLong(value(prop, "SubscriptionMaxStaleSeconds", "604800"));
+            subscriptionMaxPayloadBytes = Integer.parseInt(value(prop, "SubscriptionMaxPayloadBytes", "52428800"));
+            subscriptionHttpPort = Integer.parseInt(value(prop, "SubscriptionHttpPort", "8889"));
+            subscriptionHttpWorkers = Integer.parseInt(value(prop, "SubscriptionHttpWorkers", "4"));
+            subscriptionSocketTimeoutMs = Integer.parseInt(value(prop, "SubscriptionSocketTimeoutMs", "10000"));
+            if (subscriptionSyncEnabled && subscriptionSyncSecret.isBlank())
+                throw new IllegalStateException("启用订阅同步时必须配置 SUBSCRIPTION_SYNC_SECRET");
+        } catch (IOException ex) {
+            log.error("加载配置失败:{}", ex.getMessage());
+        }
+    }
+
+    private static String value(Properties prop, String key, String fallback) {
+        return prop.getProperty(key, fallback).trim();
+    }
+}

src/main/java/lion/CustomUtil.java

@@ -3,16 +3,16 @@ package lion;
 import cn.hutool.http.HttpRequest;
 import cn.hutool.http.HttpResponse;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
 
-import java.util.concurrent.atomic.AtomicInteger;
+import java.io.*;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketException;
 
-
+@Slf4j
-@Data
 public class CustomUtil {
 
-    public static AtomicInteger counter = new AtomicInteger();
-
     public static ObjectMapper objectMapper = new ObjectMapper();
 
     public static void notifyMe(String message) {
@@ -25,4 +25,70 @@ public class CustomUtil {
             }
         }
     }
+
+    public static int _findIdlePort(int port) {
+        for(int i=port; i<65535; i++){
+            try(ServerSocket ignored = new ServerSocket(i)){
+                return i;
+            }catch (IOException ignored) {
+            }
+        }
+        return -1;
+    }
+
+    public static String getRequestHeader(BufferedReader requestReader) throws IOException {
+        String line;
+        while ((line = requestReader.readLine()) != null) {
+            if (line.trim().isEmpty()) {
+                break;
+            }
+            if (line.startsWith("Range:")) {
+                return line.substring("Range".length() + 1).trim();
+            }
+        }
+        return null;
+    }
+
+    public static void sendErrorResponse(Socket clientSocket, String statusCode) throws IOException {
+        OutputStream responseStream = clientSocket.getOutputStream();
+        PrintWriter responseWriter = new PrintWriter(responseStream, true);
+        responseWriter.println("HTTP/1.1 " + statusCode);
+        responseWriter.println("Content-Type: text/html");
+        responseWriter.println();
+        responseWriter.println("<h1>" + statusCode + "</h1>");
+        responseStream.close();
+    }
+
+    public static void sendFileRange(Socket clientSocket, File file, long startByte, long endByte) throws IOException {
+        sendFileRange(clientSocket, file, startByte, endByte, null);
+    }
+
+    public static void sendFileRange(Socket clientSocket, File file, long startByte, long endByte, String contentDispositionFileName) throws IOException {
+        long fileLength = file.length();
+        OutputStream responseStream = clientSocket.getOutputStream();
+        PrintWriter responseWriter = new PrintWriter(responseStream, true);
+        responseWriter.println("HTTP/1.1 206 Partial Content");
+        responseWriter.println("Content-Type: application/octet-stream");
+        responseWriter.println("Accept-Ranges: bytes");
+        responseWriter.println("Content-Length: " + (endByte - startByte + 1));
+        responseWriter.println("Content-Range: bytes " + startByte + "-" + endByte + "/" + fileLength);
+        if (contentDispositionFileName != null) {
+            responseWriter.println("Content-Disposition: attachment; filename=\"" + contentDispositionFileName + "\"");
+        }
+        responseWriter.println();
+
+        try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
+            randomAccessFile.seek(startByte);
+            byte[] buffer = new byte[1024];
+            int bytesRead;
+            long bytesRemaining = endByte - startByte + 1;
+            while (bytesRemaining > 0 && (bytesRead = randomAccessFile.read(buffer, 0, (int) Math.min(buffer.length, bytesRemaining))) != -1) {
+                responseStream.write(buffer, 0, bytesRead);
+                bytesRemaining -= bytesRead;
+            }
+        } catch (SocketException ignore) {
+        } finally {
+            responseStream.close();
+        }
+    }
 }

src/main/java/lion/Domain/GalleryTask.java

@@ -6,22 +6,22 @@ import lombok.Data;
 
 @Data
 public class GalleryTask {
-    public static byte DOWNLOADING = 1;
+    public static final byte DOWNLOADING = 1;
-    public static byte DOWNLOAD_COMPLETE = 2;
+    public static final byte DOWNLOAD_COMPLETE = 2;
-    public static byte COMPRESSING = 3;
+    public static final byte COMPRESSING = 3;
-    public static byte COMPRESS_COMPLETE = 4;
+    public static final byte COMPRESS_COMPLETE = 4;
 
     @JsonInclude(JsonInclude.Include.NON_NULL)
-    private String name;
+    private volatile String name;
 
     private int gid;
 
-    private byte status;
+    private volatile byte status;
 
-    private int proceeding;
+    private volatile int proceeding;
 
     @JsonIgnore
-    private String path;
+    private volatile String path;
 
     @JsonIgnore
     public boolean is_download_complete(){

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

@@ -0,0 +1,159 @@
+package lion.Externel;
+
+import lion.Config.Config;
+import lion.CustomUtil;
+import lion.Service.SubscriptionSnapshotStore;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.*;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.concurrent.*;
+
+/** HTTP distributor for the last-known-good subscription snapshot. */
+@Slf4j
+public final class BackupSubServer implements Runnable {
+    private final SubscriptionSnapshotStore snapshotStore;
+    private final int port;
+    private final ExecutorService workers;
+
+    public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount) {
+        this.snapshotStore = Objects.requireNonNull(snapshotStore);
+        this.port = port;
+        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() {
+        try (ServerSocket serverSocket = new ServerSocket(port)) {
+            log.info("备机订阅服务监听端口 {}", port);
+            while (!Thread.currentThread().isInterrupted()) {
+                Socket socket = serverSocket.accept();
+                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());
+        } finally {
+            workers.shutdownNow();
+        }
+    }
+
+    private void handle(Socket socket) {
+        try (socket) {
+            socket.setSoTimeout(Config.subscriptionSocketTimeoutMs);
+            BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream(), StandardCharsets.US_ASCII));
+            String requestLine = reader.readLine();
+            if (requestLine == null || requestLine.length() > 2048) {
+                send(socket, 400, "Bad Request", "text/plain", new byte[0], false);
+                return;
+            }
+            String[] parts = requestLine.split(" ", 3);
+            if (parts.length != 3 || (!"GET".equals(parts[0]) && !"HEAD".equals(parts[0]))) {
+                send(socket, 405, "Method Not Allowed", "text/plain", new byte[0], false);
+                return;
+            }
+            Map<String, String> headers = readHeaders(reader);
+            if (headers == null) {
+                send(socket, 400, "Bad Request", "text/plain", new byte[0], false);
+                return;
+            }
+            String path = parts[1].split("\\?", 2)[0];
+            if ("/health/subscription".equals(path)) {
+                byte[] body = CustomUtil.objectMapper.writeValueAsBytes(snapshotStore.status());
+                send(socket, 200, "OK", "application/json; charset=utf-8", body, "HEAD".equals(parts[0]));
+                return;
+            }
+            String[] segments = path.split("/");
+            if (segments.length != 4 || !"sub".equals(segments[1]) || !("v2".equals(segments[2]) || "cat".equals(segments[2]))) {
+                send(socket, 404, "Not Found", "text/plain", new byte[0], false);
+                return;
+            }
+            String key = URLDecoder.decode(segments[3], StandardCharsets.UTF_8);
+            if (key.length() < 6 || key.length() > 512) {
+                send(socket, 404, "Not Found", "text/plain", new byte[0], false);
+                return;
+            }
+            SubscriptionSnapshotStore.Lookup lookup = snapshotStore.lookup(segments[2], key);
+            if (lookup == null) {
+                SubscriptionSnapshotStore.Status status = snapshotStore.status();
+                int code = "unavailable".equals(status.state()) || "expired".equals(status.state()) ? 503 : 404;
+                send(socket, code, code == 503 ? "Service Unavailable" : "Not Found", "text/plain", new byte[0], false);
+                return;
+            }
+            serveContent(socket, headers.get("range"), lookup.content(), "v2".equals(segments[2]), "HEAD".equals(parts[0]));
+        } catch (Exception e) {
+            log.debug("处理备机订阅请求失败: {}", e.getMessage());
+        }
+    }
+
+    private static Map<String, String> readHeaders(BufferedReader reader) throws IOException {
+        Map<String, String> headers = new HashMap<>();
+        int total = 0;
+        String line;
+        while ((line = reader.readLine()) != null) {
+            total += line.length();
+            if (total > 8192) return null;
+            if (line.isEmpty()) return headers;
+            int colon = line.indexOf(':');
+            if (colon <= 0) return null;
+            headers.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT), line.substring(colon + 1).trim());
+        }
+        return null;
+    }
+
+    private static void serveContent(Socket socket, String range, byte[] content, boolean v2, boolean head) throws IOException {
+        long start = 0;
+        long end = content.length - 1L;
+        int status = 200;
+        String reason = "OK";
+        if (range != null && range.startsWith("bytes=")) {
+            String value = range.substring(6).split(",", 2)[0];
+            String[] values = value.split("-", 2);
+            try {
+                if (values.length != 2 || values[0].isEmpty()) throw new NumberFormatException();
+                start = Long.parseLong(values[0]);
+                if (!values[1].isEmpty()) end = Long.parseLong(values[1]);
+                if (start < 0 || start > end || start >= content.length) throw new NumberFormatException();
+                end = Math.min(end, content.length - 1L);
+                status = 206;
+                reason = "Partial Content";
+            } catch (NumberFormatException e) {
+                send(socket, 416, "Range Not Satisfiable", v2 ? "text/plain" : "text/yaml", new byte[0], head);
+                return;
+            }
+        }
+        byte[] body = Arrays.copyOfRange(content, (int) start, (int) end + 1);
+        send(socket, status, reason, v2 ? "text/plain; charset=utf-8" : "text/yaml; charset=utf-8", body, head,
+                status == 206 ? "bytes " + start + "-" + end + "/" + content.length : null);
+    }
+
+    private static void send(Socket socket, int status, String reason, String type, byte[] body, boolean head) throws IOException {
+        send(socket, status, reason, type, body, head, null);
+    }
+
+    private static void send(Socket socket, int status, String reason, String type, byte[] body, boolean head, String range) throws IOException {
+        OutputStream output = socket.getOutputStream();
+        StringBuilder header = new StringBuilder()
+                .append("HTTP/1.1 ").append(status).append(' ').append(reason).append("\r\n")
+                .append("Content-Type: ").append(type).append("\r\n")
+                .append("Content-Length: ").append(body.length).append("\r\n")
+                .append("Accept-Ranges: bytes\r\n")
+                .append("Connection: close\r\n");
+        if (range != null) header.append("Content-Range: ").append(range).append("\r\n");
+        header.append("\r\n");
+        output.write(header.toString().getBytes(StandardCharsets.US_ASCII));
+        if (!head) output.write(body);
+        output.flush();
+    }
+}

src/main/java/lion/Extranel/AESUtils.java

@@ -1,26 +0,0 @@
-package lion.Extranel;
-
-import javax.crypto.Cipher;
-import javax.crypto.spec.SecretKeySpec;
-import java.security.Key;
-
-public class AESUtils {
-    private static final String ALGORITHM = "AES";
-    private static final String TRANSFORMATION = "AES/ECB/PKCS5Padding";
-
-    private static final byte[] keyBytes = "ThisIsA128BitKey".getBytes();
-
-    public static byte[] encrypt(byte[] data) throws Exception {
-        Key key = new SecretKeySpec(keyBytes, ALGORITHM);
-        Cipher cipher = Cipher.getInstance(TRANSFORMATION);
-        cipher.init(Cipher.ENCRYPT_MODE, key);
-        return cipher.doFinal(data);
-    }
-
-    public static byte[] decrypt(byte[] encryptedData) throws Exception {
-        Key key = new SecretKeySpec(keyBytes, ALGORITHM);
-        Cipher cipher = Cipher.getInstance(TRANSFORMATION);
-        cipher.init(Cipher.DECRYPT_MODE, key);
-        return cipher.doFinal(encryptedData);
-    }
-}

src/main/java/lion/Extranel/Server.java

@@ -1,65 +0,0 @@
-package lion.Extranel;
-
-import com.fasterxml.jackson.databind.ObjectMapper;
-import lion.CustomUtil;
-import lombok.extern.slf4j.Slf4j;
-
-import java.io.BufferedInputStream;
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.net.URI;
-import java.net.URL;
-import java.util.HashMap;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-@Slf4j
-public class Server {
-
-    ExecutorService thread_pool;
-
-    ObjectMapper objectMapper;
-
-    public static void main(String[] args) {
-        new Server();
-    }
-
-    public Server(){
-        log.info("开始监听:55555");
-        objectMapper = CustomUtil.objectMapper;
-        thread_pool = Executors.newFixedThreadPool(2);
-        thread_pool.submit(() -> {
-            try(ServerSocket server = new ServerSocket(55555)){
-                server.setSoTimeout(1000000);
-                while (true){
-                    Socket socket = server.accept();
-                    thread_pool.submit(() -> handleSocket(socket));
-                }
-            }catch (IOException e){
-                log.error(e.getMessage());
-            }
-        });
-    }
-
-
-    public void handleSocket(Socket socket){
-        try{
-            BufferedInputStream inputStream = new BufferedInputStream(socket.getInputStream());
-            Thread.sleep(500);
-            byte[] buf = new byte[inputStream.available()];
-            inputStream.read(buf);
-            byte[] bytes = AESUtils.decrypt(buf);
-
-            HashMap<String, String> map = (HashMap<String, String>) objectMapper.readValue(bytes, HashMap.class);
-            URL url = new URI(map.get("path")).toURL();
-            log.info("处理反代 ip: {},路径: {}", socket.getInetAddress().getHostAddress(), map.get("path"));
-            bytes = url.openConnection().getInputStream().readAllBytes();
-            socket.getOutputStream().write(AESUtils.encrypt(bytes));
-            socket.getOutputStream().flush();
-            socket.getOutputStream().close();
-        } catch (Exception e) {
-           e.printStackTrace();
-        }
-    }
-}

src/main/java/lion/Main.java

@@ -1,7 +1,9 @@
 package lion;
 
 import io.netty.bootstrap.Bootstrap;
-import lion.Extranel.Server;
+import lion.Config.Config;
+import lion.Externel.BackupSubServer;
+import lion.Service.SubscriptionSnapshotStore;
 import lombok.extern.slf4j.Slf4j;
 
 @Slf4j
@@ -9,9 +11,15 @@ public class Main {
 
     public static void main(String[] args) {
         boot();
+        Config.loadConfig();
+        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();
-        new Thread(() -> Server.main(null)).start();
+        new storageNode(snapshotStore);
-        new storageNode();
     }
 
     public static void boot(){

src/main/java/lion/Message/AbstractMessage.java

@@ -14,6 +14,10 @@ public class AbstractMessage {
 
     public static final byte MAINTAIN_MESSAGE = 7;
 
+    public static final byte AVAILABLE_CHECK_MESSAGE = 8;
+
+    public static final byte SUBSCRIPTION_SNAPSHOT_MESSAGE = 9;
+
     public byte messageType;
 
     public int messageId;

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

@@ -0,0 +1,11 @@
+package lion.Message.Main;
+
+import lion.Message.AbstractMessage;
+import lombok.Data;
+
+@Data
+public class AvailableCheckMessage extends AbstractMessage {
+    {
+        messageType = AVAILABLE_CHECK_MESSAGE;
+    }
+}

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

@@ -0,0 +1,16 @@
+package lion.Message.Main;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class SubscriptionAccountSnapshot {
+    private Integer accountId;
+    private boolean enabled;
+    private boolean filterHighMultiplier;
+    private String v2ContentBase64;
+    private String v2Sha256;
+    private String clashContentBase64;
+    private String clashSha256;
+}

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

@@ -0,0 +1,11 @@
+package lion.Message.Main;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@NoArgsConstructor
+public class SubscriptionBindingSnapshot {
+    private String publicKeySha256;
+    private Integer accountId;
+}

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

@@ -0,0 +1,22 @@
+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;
+    }
+
+    private int schemaVersion;
+    private String revision;
+    private long generatedAt;
+    private String payloadBase64;
+    private String payloadSha256;
+    private String signature;
+}

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

@@ -0,0 +1,15 @@
+package lion.Message.Main;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+@Data
+@NoArgsConstructor
+public class SubscriptionSnapshotPayload {
+    private int schemaVersion;
+    private List<SubscriptionAccountSnapshot> accounts = new ArrayList<>();
+    private List<SubscriptionBindingSnapshot> bindings = new ArrayList<>();
+}

src/main/java/lion/Message/MessageCodec.java

@@ -25,9 +25,13 @@ public class MessageCodec extends ByteToMessageCodec<AbstractMessage> {
     protected void encode(ChannelHandlerContext channelHandlerContext, AbstractMessage abstractMessage, ByteBuf byteBuf) {
         byteBuf.writeByte(abstractMessage.messageType);
 
-        byte[] bytes = objectMapper.valueToTree(abstractMessage).toString().getBytes(StandardCharsets.UTF_8);
+        try {
+            byte[] bytes = objectMapper.writeValueAsBytes(abstractMessage);
             byteBuf.writeInt(bytes.length);
             byteBuf.writeBytes(bytes);
+        } catch (Exception e) {
+            log.error("序列化消息失败:{}", e.getMessage());
+        }
     }
 
     @Override
@@ -45,6 +49,8 @@ public class MessageCodec extends ByteToMessageCodec<AbstractMessage> {
             case AbstractMessage.DELETE_GALLERY_MESSAGE -> objectMapper.readValue(metadata, DeleteGalleryMessage.class);
             case AbstractMessage.IDENTITY_MESSAGE -> objectMapper.readValue(metadata, IdentityMessage.class);
             case AbstractMessage.MAINTAIN_MESSAGE -> objectMapper.readValue(metadata, MaintainMessage.class);
+            case AbstractMessage.AVAILABLE_CHECK_MESSAGE -> objectMapper.readValue(metadata, AvailableCheckMessage.class);
+            case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> objectMapper.readValue(metadata, SubscriptionSnapshotMessage.class);
             default -> null;
         };
 

src/main/java/lion/MultiThreadedHTTPServer.java

@@ -10,10 +10,10 @@ import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+
 @Slf4j
 public class MultiThreadedHTTPServer {
     private static final int PORT = 8888;
-    private static final int BUFFER_SIZE = 1024;
 
     public static void main(String[] args) {
         ExecutorService threadPool = Executors.newCachedThreadPool();
@@ -30,13 +30,11 @@ public class MultiThreadedHTTPServer {
                 String ip = clientSocket.getInetAddress().getHostAddress();
                 if(ip.equals(real_ip)){
                     log.info("Client connected");
-                    // 线程池处理下载请求
                     threadPool.submit(() -> handleClientRequest(clientSocket));
                 }else{
                     log.info("unknown ip: " + ip);
                     clientSocket.close();
                 }
-
             }
         } catch (IOException e) {
             log.error("处理http请求时出错,IP:{},ERROR:{}", real_ip, e.getMessage());
@@ -49,6 +47,11 @@ public class MultiThreadedHTTPServer {
             BufferedReader requestReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
             String requestLine = requestReader.readLine();
 
+            if (requestLine == null) {
+                clientSocket.close();
+                return;
+            }
+
             // Parse the request line to get the method and path
             String[] requestParts = requestLine.split(" ");
             String method = requestParts[0];
@@ -66,38 +69,23 @@ public class MultiThreadedHTTPServer {
                     }
                     else {
                         String filePath = "/root/gallery/gallery";
+                        String gid = paramMap.get("gid");
+                        file = gid == null ? null : findGalleryZipByGid(new File(filePath), gid);
+
+                        //兼容没有gid参数的旧下载链接，再尝试按链接中的文件名查找
+                        if(file == null){
                             String path = URLDecoder.decode(requestParts[1].split("\\?")[0], StandardCharsets.UTF_8);
-                        if(!path.contains(".")){
+                            if(path.contains(".")){
-                            file = new File("/root/abc");
-                        }else {
                                 String name = path.substring(0, path.lastIndexOf('.'));
                                 name = filePath + name + "/" + name + ".zip";
                                 file = new File(name);
-
+                            }else{
-                            //该文件不存在
-                            if(!file.isFile()){
-                                String gid = paramMap.get("gid");
-
-                                //文件不存在的情况下gid也不存在，直接404
-                                if(gid == null)
                                 file = new File("/root/abc");
-
-                                //gid存在的情况下尝试查找对应的文件
-                                else {
-                                    File[] galleryDirectories = (new File(filePath)).listFiles();
-
-                                    assert galleryDirectories != null;
-                                    for (File galleryDirectory : galleryDirectories)
-                                        if (galleryDirectory.getName().contains(gid)) {
-                                            file = new File(galleryDirectory.getAbsolutePath(), galleryDirectory.getName() + ".zip");
-                                            break;
-                                        }
-                                }
                             }
                         }
                     }
                 else{
-                    sendErrorResponse(clientSocket, "403 Forbidden");
+                    CustomUtil.sendErrorResponse(clientSocket, "403 Forbidden");
                     return;
                 }
                 fileName = file.getName();
@@ -110,7 +98,7 @@ public class MultiThreadedHTTPServer {
                     // Get the range information for resuming download
                     long startByte = 0;
                     long endByte = fileLength - 1;
-                    String rangeHeader = getRequestHeader(requestReader);
+                    String rangeHeader = CustomUtil.getRequestHeader(requestReader);
                     if (rangeHeader != null && rangeHeader.startsWith("bytes=")) {
                         String[] rangeValues = rangeHeader.substring(6).split("-");
                         startByte = Long.parseLong(rangeValues[0]);
@@ -119,40 +107,14 @@ public class MultiThreadedHTTPServer {
                         }
                     }
 
-                    // Send the HTTP response headers
+                    CustomUtil.sendFileRange(clientSocket, file, startByte, endByte, fileName);
-                    OutputStream responseStream = clientSocket.getOutputStream();
-                    PrintWriter responseWriter = new PrintWriter(responseStream, true);
-                    responseWriter.println("HTTP/1.1 206 Partial Content");
-                    responseWriter.println("Content-Type: application/octet-stream");
-                    responseWriter.println("Accept-Ranges: bytes");
-                    responseWriter.println("Content-Length: " + (endByte - startByte + 1));
-                    responseWriter.println("Content-Range: bytes " + startByte + "-" + endByte + "/" + fileLength);
-                    responseWriter.println("Content-Disposition: attachment; filename=\"" + fileName + "\"");
-                    responseWriter.println();
-
-                    // Send the file content
-                    try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
-                        randomAccessFile.seek(startByte);
-                        byte[] buffer = new byte[BUFFER_SIZE];
-                        int bytesRead;
-                        long bytesRemaining = endByte - startByte + 1;
-                        while (bytesRemaining > 0 && (bytesRead = randomAccessFile.read(buffer, 0, (int) Math.min(buffer.length, bytesRemaining))) != -1) {
-                            responseStream.write(buffer, 0, bytesRead);
-                            bytesRemaining -= bytesRead;
-                        }
-                    }catch (SocketException ignore){
-
-                    }
-
-                    // Close the response output stream
-                    responseStream.close();
                 } else {
                     // File not found or not readable, send 404 response
-                    sendErrorResponse(clientSocket, "404 Not Found");
+                    CustomUtil.sendErrorResponse(clientSocket, "404 Not Found");
                 }
             } else {
                 // Non-GET requests, send 501 response
-                sendErrorResponse(clientSocket, "501 Not Implemented");
+                CustomUtil.sendErrorResponse(clientSocket, "501 Not Implemented");
             }
 
             // Close the request reader and client socket
@@ -163,16 +125,24 @@ public class MultiThreadedHTTPServer {
         }
     }
 
-    private static String getRequestHeader(BufferedReader requestReader) throws IOException {
+    private static File findGalleryZipByGid(File galleryRoot, String gid){
-        String line;
+        File[] galleryDirectories = galleryRoot.listFiles(File::isDirectory);
-        while ((line = requestReader.readLine()) != null) {
+        if(galleryDirectories == null)
-            if (line.trim().isEmpty()) {
+            return null;
-                break;
-            }
 
-            if (line.startsWith("Range" + ":")) {
+        String marker = "[" + gid;
-                return line.substring("Range".length() + 1).trim();
+        for(File galleryDirectory : galleryDirectories){
-            }
+            String name = galleryDirectory.getName();
+            int markerIndex = name.lastIndexOf(marker);
+            if(markerIndex < 0)
+                continue;
+            int suffixIndex = markerIndex + marker.length();
+            if(suffixIndex >= name.length() || (name.charAt(suffixIndex) != ']' && name.charAt(suffixIndex) != '-'))
+                continue;
+
+            File zip = new File(galleryDirectory, name + ".zip");
+            if(zip.isFile())
+                return zip;
         }
         return null;
     }
@@ -206,14 +176,4 @@ public class MultiThreadedHTTPServer {
         }
         return queryParams;
     }
-
-    private static void sendErrorResponse(Socket clientSocket, String statusCode) throws IOException {
-        OutputStream responseStream = clientSocket.getOutputStream();
-        PrintWriter responseWriter = new PrintWriter(responseStream, true);
-        responseWriter.println("HTTP/1.1 " + statusCode);
-        responseWriter.println("Content-Type: text/html");
-        responseWriter.println();
-        responseWriter.println("<h1>" + statusCode + "</h1>");
-        responseStream.close();
-    }
 }

src/main/java/lion/Service/DownloadCheckService.java

@@ -1,26 +1,22 @@
 package lion.Service;
 
-import io.netty.channel.Channel;
-import io.netty.channel.DefaultEventLoop;
-import io.netty.channel.EventLoop;
-import io.netty.util.concurrent.Promise;
 import lion.CustomUtil;
 import lion.Domain.GalleryTask;
 import cn.hutool.core.io.FileUtil;
 import cn.hutool.core.util.ZipUtil;
-import lion.Message.AbstractMessage;
-import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.*;
 import java.util.*;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
+import java.nio.file.*;
-
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.zip.CRC32;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipFile;
 
 @Slf4j
-@Data
 public class DownloadCheckService {
     Map<Integer, GalleryTask> queue;
 
@@ -30,21 +26,22 @@ public class DownloadCheckService {
 
     ScheduledThreadPoolExecutor convert_thread;
 
-    ArrayList<GalleryTask> compress_queue;
+    final ArrayList<GalleryTask> compress_queue;
 
-    Channel node;
+    final Map<Integer, Long> retryAfter = new ConcurrentHashMap<>();
 
-    HashMap<Integer, Promise<AbstractMessage>> promises;
+    public DownloadCheckService(Map<Integer, GalleryTask> queue){
+        this(queue, true);
+    }
 
-    EventLoop eventLoop;
+    // Tests use temporary directories and invoke scans explicitly.
-
+    DownloadCheckService(Map<Integer, GalleryTask> queue, boolean startScheduler){
-    public DownloadCheckService(Map<Integer, GalleryTask> queue, HashMap<Integer, Promise<AbstractMessage>> promises){
         this.queue = queue;
-        this.promises = promises;
-        eventLoop = new DefaultEventLoop();
         compress_queue = new ArrayList<>(0);
+        if (startScheduler) {
             convert_thread = new ScheduledThreadPoolExecutor(1);
-        convert_thread.scheduleAtFixedRate(this::compress, 0, 5, TimeUnit.SECONDS);
+            convert_thread.scheduleWithFixedDelay(this::compress, 0, 5, TimeUnit.SECONDS);
+        }
     }
 
     public boolean downloadCheck(){
@@ -72,7 +69,7 @@ public class DownloadCheckService {
 
             while(fileIterator.hasNext()){
                 File file = fileIterator.next();
-                if(!file.getName().contains(String.valueOf(galleryTask.getGid())))
+                if(!file.isDirectory() || !matchesGid(file.getName(), galleryTask.getGid()))
                     continue;
 
                 galleryTask.setStatus(GalleryTask.DOWNLOADING);
@@ -98,10 +95,13 @@ public class DownloadCheckService {
 
         //压缩队列
         for(GalleryTask galleryTask: queue.values())
-            if (galleryTask.is_download_complete()) {
+            if (galleryTask.is_download_complete()
+                    && System.currentTimeMillis() >= retryAfter.getOrDefault(galleryTask.getGid(), 0L)) {
                 galleryTask.setStatus(GalleryTask.COMPRESSING);
+                synchronized (compress_queue) {
                     compress_queue.add(galleryTask);
                 }
+            }
 
         return result;
     }
@@ -110,65 +110,163 @@ public class DownloadCheckService {
      * 压缩线程：将压缩队列的任务复制一份，进行转换
      */
     public void compress() {
-        if(compress_queue.isEmpty())
+        ArrayList<GalleryTask> galleryTasks;
+        synchronized (compress_queue) {
+            if (compress_queue.isEmpty())
                 return;
-        ReentrantLock reentrantLock = new ReentrantLock();
+            galleryTasks = new ArrayList<>(compress_queue);
-        reentrantLock.lock();
-        ArrayList<GalleryTask> galleryTasks = new ArrayList<>(compress_queue);
             compress_queue.clear();
-        reentrantLock.unlock();
+        }
         for (GalleryTask galleryTask : galleryTasks) {
+            Path temporary = null;
             try {
                 log.info("开始压缩:{}", galleryTask.getName());
-                    //创建文件夹
+                Path directory = Paths.get(storagePath, galleryTask.getName());
-                    File file = new File(storagePath + galleryTask.getName());
+                Files.createDirectories(directory);
-                    if (file.isDirectory() || file.mkdirs()) {
+                Path archive = directory.resolve(galleryTask.getName() + ".zip");
-                        log.info("{}文件夹创建成功", galleryTask.getName());
+                temporary = Files.createTempFile(directory, ".compress-", ".zip.part");
-                    } else {
+                ZipUtil.zip(galleryTask.getPath(), temporary.toString());
-                        log.error("{}文件夹创建失败", galleryTask.getName());
+                if (!isValidArchive(temporary.toFile()))
-                        continue;
+                    throw new IOException("压缩包校验失败");
+                // Publish only a closed, verified archive. A crash leaves a .part file.
+                try {
+                    Files.move(temporary, archive, StandardCopyOption.ATOMIC_MOVE, StandardCopyOption.REPLACE_EXISTING);
+                } catch (AtomicMoveNotSupportedException e) {
+                    Files.move(temporary, archive, StandardCopyOption.REPLACE_EXISTING);
                 }
-
+                temporary = null;
-                    //生成压缩包
-                    ZipUtil.zip(galleryTask.getPath(), storagePath + galleryTask.getName() + "/" + galleryTask.getName() + ".zip");
-                    log.info("{}压缩完成", galleryTask.getName());
-
-                    FileUtil.del(galleryTask.getPath());
                 galleryTask.setStatus(GalleryTask.COMPRESS_COMPLETE);
-                } catch (Exception e){
+                retryAfter.remove(galleryTask.getGid());
-                    log.error("{}压缩失败:{}", galleryTask, e.getMessage());
+                if (!FileUtil.del(galleryTask.getPath()))
+                    log.warn("压缩已完成，但源目录清理失败: {}", galleryTask.getPath());
+                log.info("{}压缩完成", galleryTask.getName());
+            } catch (Exception e) {
+                // Keep the source and restore an existing, retryable protocol state.
+                if (!galleryTask.is_compress_complete()) {
+                    retryAfter.put(galleryTask.getGid(), System.currentTimeMillis() + 30_000);
+                    galleryTask.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
+                }
+                log.error("{}压缩或清理失败，源文件保留，稍后可重试", galleryTask.getName(), e);
+            } finally {
+                if (temporary != null) {
+                    try { Files.deleteIfExists(temporary); }
+                    catch (IOException e) { log.warn("清理压缩临时文件失败: {}", temporary, e); }
+                }
             }
         }
     }
 
-    /**
+    static boolean isValidArchive(File file) {
-     * 检查改任务是否为已完成任务，如已完成则返回true，若未完成则加入队列
+        if (!file.isFile())
-     * @return true if compress complete, false otherwise
+            return false;
-     */
+        try (ZipFile zip = new ZipFile(file)) {
-    public boolean addToQueue(GalleryTask galleryTask){
+            if (zip.size() == 0)
-        //是否含有名字，进行中任务一般有名字，没有名字则肯定为初始任务，存在名字至少在下载路径出现过
+                return false;
-        if(galleryTask.getName() == null || !galleryTask.getName().isEmpty()){
+            byte[] buffer = new byte[8192];
-            queue.putIfAbsent(galleryTask.getGid(), galleryTask);
+            Enumeration<? extends ZipEntry> entries = zip.entries();
+            while (entries.hasMoreElements()) {
+                ZipEntry entry = entries.nextElement();
+                if (entry.isDirectory())
+                    continue;
+                CRC32 crc = new CRC32();
+                long size = 0;
+                try (InputStream input = zip.getInputStream(entry)) {
+                    int count;
+                    while ((count = input.read(buffer)) != -1) {
+                        crc.update(buffer, 0, count);
+                        size += count;
+                    }
+                }
+                if (size != entry.getSize() || crc.getValue() != entry.getCrc())
                     return false;
             }
-
+            return true;
-        //查询hah下载路径中，是否存在该任务下载路径，存在则为下载中或下载完成任务，加入队列
+        } catch (IOException e) {
-        if(new File(downloadPath + galleryTask.getGid()).isDirectory()){
-            queue.putIfAbsent(galleryTask.getGid(), galleryTask);
             return false;
         }
+    }
 
-
+    public GalleryTask addToQueue(GalleryTask galleryTask){
-        //查询存放路径中是否含有该任务的压缩包，存在则为下载完成任务
+        GalleryTask active = queue.get(galleryTask.getGid());
-        if(new File(storagePath + galleryTask.getName() + "/" + galleryTask.getName() + ".zip").exists()){
+        if (active != null && active.is_compressing())
+            return active;
+        retryAfter.remove(galleryTask.getGid());
+        // A reconnect can resend a task whose name is stale (for example, the
+        // downloader appended a resolution suffix). Resolve completed archives
+        // by gid first, because the gid is stable while the directory name is not.
+        File storedDirectory = findStoredDirectoryByGid(galleryTask.getGid());
+        if(storedDirectory != null){
+            queue.remove(galleryTask.getGid());
+            galleryTask.setName(storedDirectory.getName());
             galleryTask.setStatus(GalleryTask.COMPRESS_COMPLETE);
             CustomUtil.notifyMe(String.format("任务:%s在添加时已下载完成，更新任务状态", galleryTask.getName()));
-            return true;
+            return galleryTask;
         }
 
+        // If the task is still downloading, retain the actual directory name so
+        // subsequent progress and compression use the same identity.
+        File downloadingDirectory = findDirectoryByGid(new File(downloadPath), galleryTask.getGid());
+        if(downloadingDirectory != null){
+            GalleryTask queuedTask = queue.get(galleryTask.getGid());
+            if(queuedTask == null){
+                refreshDownloadingTask(galleryTask, downloadingDirectory);
+                queue.put(galleryTask.getGid(), galleryTask);
+                return galleryTask;
+            }
+            if(!queuedTask.is_compressing() && !queuedTask.is_compress_complete())
+                refreshDownloadingTask(queuedTask, downloadingDirectory);
+            return queuedTask;
+        }
 
-        //异常情况，发送通知
+        GalleryTask queuedTask = queue.putIfAbsent(galleryTask.getGid(), galleryTask);
-        CustomUtil.notifyMe(String.format("任务:%s存在名字，但是下载路径为空且不存在压缩包", galleryTask.getName() ));
+        return queuedTask == null ? galleryTask : queuedTask;
+    }
+
+    private void refreshDownloadingTask(GalleryTask galleryTask, File downloadingDirectory){
+        galleryTask.setName(downloadingDirectory.getName());
+        File[] pages = downloadingDirectory.listFiles((dir, name) -> !name.equals("galleryinfo.txt"));
+        galleryTask.setProceeding(pages == null ? 0 : pages.length);
+        if(new File(downloadingDirectory, "galleryinfo.txt").isFile()){
+            galleryTask.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
+            galleryTask.setPath(downloadingDirectory.getPath());
+        }else{
+            galleryTask.setStatus(GalleryTask.DOWNLOADING);
+        }
+    }
+
+    private File findStoredDirectoryByGid(int gid){
+        File storageDirectory = new File(storagePath);
+        File[] directories = storageDirectory.listFiles(File::isDirectory);
+        if(directories == null)
+            return null;
+
+        for(File directory : directories){
+            if(matchesGid(directory.getName(), gid)
+                    && isValidArchive(new File(directory, directory.getName() + ".zip")))
+                return directory;
+        }
+        return null;
+    }
+
+    private File findDirectoryByGid(File parentDirectory, int gid){
+        File[] directories = parentDirectory.listFiles(File::isDirectory);
+        if(directories == null)
+            return null;
+
+        for(File directory : directories)
+            if(matchesGid(directory.getName(), gid))
+                return directory;
+        return null;
+    }
+
+    private boolean matchesGid(String name, int gid){
+        String gidMarker = "[" + gid;
+        int markerIndex = name.lastIndexOf(gidMarker);
+        if(markerIndex < 0)
             return false;
+
+        int suffixIndex = markerIndex + gidMarker.length();
+        return suffixIndex < name.length()
+                && (name.charAt(suffixIndex) == ']' || name.charAt(suffixIndex) == '-');
     }
 }

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

@@ -0,0 +1,329 @@
+package lion.Service;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lion.CustomUtil;
+import lion.Message.Main.SubscriptionAccountSnapshot;
+import lion.Message.Main.SubscriptionBindingSnapshot;
+import lion.Message.Main.SubscriptionSnapshotMessage;
+import lion.Message.Main.SubscriptionSnapshotPayload;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.*;
+import java.security.MessageDigest;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.zip.GZIPInputStream;
+
+/**
+ * Last-known-good subscription data used by the standby HTTP server.
+ * The store never downloads upstream subscriptions and never stores upstream keys.
+ */
+@Slf4j
+public final class SubscriptionSnapshotStore {
+    public static final byte APPLY_SUCCESS = 0;
+    public static final byte APPLY_INVALID = 1;
+    public static final byte APPLY_IO_ERROR = 2;
+    public static final byte APPLY_OLD = 3;
+
+    private final Path root;
+    private final byte[] syncSecret;
+    private final long maxStaleMillis;
+    private final int maxPayloadBytes;
+    private final ObjectMapper objectMapper = CustomUtil.objectMapper;
+    private final AtomicReference<Snapshot> current = new AtomicReference<>();
+
+    public SubscriptionSnapshotStore(Path root, String syncSecret, long maxStaleSeconds, int maxPayloadBytes) {
+        this.root = Objects.requireNonNull(root);
+        this.syncSecret = syncSecret == null ? new byte[0] : syncSecret.getBytes(StandardCharsets.UTF_8);
+        this.maxStaleMillis = Math.max(0, maxStaleSeconds) * 1000L;
+        this.maxPayloadBytes = maxPayloadBytes;
+    }
+
+    public void load() {
+        List<String> candidates = new ArrayList<>();
+        try {
+            Path pointer = root.resolve("current-revision");
+            if (Files.isRegularFile(pointer)) {
+                String revision = Files.readString(pointer, StandardCharsets.UTF_8).trim();
+                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);
+                }
+            }
+        } catch (Exception e) {
+            log.warn("扫描订阅快照失败: {}", 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 非法");
+            if (syncSecret.length == 0)
+                return new ApplyResult(APPLY_INVALID, "同步密钥未配置");
+
+            byte[] compressed = decodeBase64(message.getPayloadBase64(), maxPayloadBytes);
+            if (!constantEquals(message.getPayloadSha256(), sha256(compressed)))
+                return new ApplyResult(APPLY_INVALID, "payload SHA-256 校验失败");
+            String signatureInput = message.getSchemaVersion() + "\n" + message.getRevision() + "\n"
+                    + message.getGeneratedAt() + "\n" + message.getPayloadSha256();
+            if (!constantEquals(message.getSignature(), hmac(signatureInput.getBytes(StandardCharsets.UTF_8))))
+                return new ApplyResult(APPLY_INVALID, "快照签名校验失败");
+
+            byte[] payloadBytes = gunzip(compressed, maxPayloadBytes);
+            if (!constantEquals(message.getRevision(), sha256(payloadBytes)))
+                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);
+            staging = snapshots.resolve(".staging-" + message.getRevision());
+            deleteRecursively(staging);
+            Files.createDirectories(staging.resolve("accounts"));
+            for (AccountFiles account : data.accounts.values()) {
+                Path dir = staging.resolve("accounts").resolve(String.valueOf(account.accountId()));
+                Files.createDirectories(dir);
+                Files.write(dir.resolve("v2ray.txt"), account.v2(), StandardOpenOption.CREATE_NEW);
+                Files.write(dir.resolve("clash.yaml"), account.clash(), StandardOpenOption.CREATE_NEW);
+            }
+            Map<String, Object> manifest = new LinkedHashMap<>();
+            manifest.put("revision", message.getRevision());
+            manifest.put("generatedAt", message.getGeneratedAt());
+            // 保留主站签名对应的原始 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);
+            writePointer(message.getRevision());
+
+            Snapshot snapshot = new Snapshot(message.getRevision(), message.getGeneratedAt(), data.byKeyHash,
+                    data.accounts.size(), data.bindingCount);
+            current.set(snapshot);
+            cleanupOldSnapshots(message.getRevision());
+            return new ApplyResult(APPLY_SUCCESS, "同步成功");
+        } 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))
+            return null;
+        AccountFiles account = snapshot.byKeyHash().get(sha256(publicKey.getBytes(StandardCharsets.UTF_8)));
+        if (account == null)
+            return null;
+        return new Lookup("v2".equals(client) ? account.v2() : account.clash(), snapshot.revision(), snapshot.generatedAt());
+    }
+
+    public Status status() {
+        Snapshot snapshot = current.get();
+        if (snapshot == null)
+            return new Status("unavailable", null, 0, 0, 0);
+        long age = Math.max(0, System.currentTimeMillis() - snapshot.generatedAt());
+        boolean expired = snapshot.expired(System.currentTimeMillis(), maxStaleMillis);
+        return new Status(expired ? "expired" : "ready", snapshot.revision(), snapshot.accountCount(), snapshot.bindingCount(), age);
+    }
+
+    private Snapshot loadSnapshot(Path directory) throws IOException {
+        JsonNode manifest = objectMapper.readTree(Files.readAllBytes(directory.resolve("manifest.json")));
+        String revision = manifest.path("revision").asText();
+        long generatedAt = manifest.path("generatedAt").asLong(0);
+        byte[] payloadBytes = decodeBase64(manifest.path("payloadBase64").asText(null), maxPayloadBytes);
+        SubscriptionSnapshotPayload payload = objectMapper.readValue(payloadBytes, SubscriptionSnapshotPayload.class);
+        SnapshotData data = validatePayload(payload);
+        if (!isRevision(revision) || !constantEquals(revision, sha256(payloadBytes)))
+            throw new IOException("快照 manifest revision 校验失败");
+        Map<Integer, AccountFiles> accounts = new HashMap<>();
+        for (SubscriptionAccountSnapshot account : payload.getAccounts()) {
+            Path dir = directory.resolve("accounts").resolve(String.valueOf(account.getAccountId()));
+            byte[] v2 = Files.readAllBytes(dir.resolve("v2ray.txt"));
+            byte[] clash = Files.readAllBytes(dir.resolve("clash.yaml"));
+            if (!constantEquals(account.getV2Sha256(), sha256(v2)) || !constantEquals(account.getClashSha256(), sha256(clash)))
+                throw new IOException("缓存文件校验失败");
+            accounts.put(account.getAccountId(), new AccountFiles(account.getAccountId(), v2, clash));
+        }
+        Map<String, AccountFiles> byKey = new HashMap<>();
+        for (SubscriptionBindingSnapshot binding : payload.getBindings())
+            byKey.put(binding.getPublicKeySha256(), accounts.get(binding.getAccountId()));
+        return new Snapshot(revision, generatedAt, byKey, accounts.size(), payload.getBindings().size());
+    }
+
+    private SnapshotData validatePayload(SubscriptionSnapshotPayload payload) throws IOException {
+        if (payload == null || payload.getSchemaVersion() != 1 || payload.getAccounts() == null || payload.getBindings() == null)
+            throw new IOException("payload 版本或字段非法");
+        if (payload.getAccounts().size() > 100 || payload.getBindings().size() > 10000)
+            throw new IOException("快照条目数量超限");
+        Map<Integer, AccountFiles> accounts = new HashMap<>();
+        for (SubscriptionAccountSnapshot account : payload.getAccounts()) {
+            if (account == null || account.getAccountId() == null || account.getAccountId() <= 0 || !account.isEnabled()
+                    || !isBase64Sha(account.getV2Sha256()) || !isBase64Sha(account.getClashSha256()))
+                throw new IOException("账号字段非法");
+            byte[] v2 = decodeBase64(account.getV2ContentBase64(), maxPayloadBytes);
+            byte[] clash = decodeBase64(account.getClashContentBase64(), maxPayloadBytes);
+            if (!constantEquals(account.getV2Sha256(), sha256(v2)) || !constantEquals(account.getClashSha256(), sha256(clash)))
+                throw new IOException("账号缓存 SHA-256 校验失败");
+            if (accounts.put(account.getAccountId(), new AccountFiles(account.getAccountId(), v2, clash)) != null)
+                throw new IOException("账号 ID 重复");
+        }
+        Map<String, AccountFiles> byKey = new HashMap<>();
+        for (SubscriptionBindingSnapshot binding : payload.getBindings()) {
+            if (binding == null || !isBase64Sha(binding.getPublicKeySha256()) || !accounts.containsKey(binding.getAccountId()))
+                throw new IOException("绑定字段或账号引用非法");
+            if (byKey.put(binding.getPublicKeySha256(), accounts.get(binding.getAccountId())) != null)
+                throw new IOException("公开 Key Hash 重复");
+        }
+        return new SnapshotData(accounts, byKey, payload.getBindings().size());
+    }
+
+    private void cleanupOldSnapshots(String currentRevision) {
+        try {
+            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-"))
+                        .sorted(Comparator.comparingLong(SubscriptionSnapshotStore::lastModified).reversed())
+                        .toList();
+            }
+            dirs.stream().filter(p -> !p.getFileName().toString().equals(currentRevision))
+                    .skip(1).forEach(p -> {
+                        try { deleteRecursively(p); } catch (IOException e) { log.warn("清理旧订阅快照失败: {}", p); }
+                    });
+        } catch (IOException e) {
+            log.warn("扫描旧订阅快照失败: {}", e.getMessage());
+        }
+    }
+
+    private static byte[] gunzip(byte[] compressed, int limit) throws IOException {
+        try (GZIPInputStream input = new GZIPInputStream(new ByteArrayInputStream(compressed))) {
+            byte[] buffer = new byte[8192];
+            var output = new java.io.ByteArrayOutputStream();
+            int total = 0, read;
+            while ((read = input.read(buffer)) != -1) {
+                total += read;
+                if (total > limit) throw new IOException("解压后 payload 超限");
+                output.write(buffer, 0, read);
+            }
+            return output.toByteArray();
+        }
+    }
+
+    private String hmac(byte[] input) throws Exception {
+        Mac mac = Mac.getInstance("HmacSHA256");
+        mac.init(new SecretKeySpec(syncSecret, "HmacSHA256"));
+        return hex(mac.doFinal(input));
+    }
+
+    private static byte[] decodeBase64(String value, int maxBytes) throws IOException {
+        if (value == null || value.length() > maxBytes * 2L)
+            throw new IOException("Base64 数据超限");
+        try {
+            byte[] result = Base64.getDecoder().decode(value);
+            if (result.length > maxBytes) throw new IOException("数据超限");
+            return result;
+        } catch (IllegalArgumentException e) {
+            throw new IOException("Base64 数据非法");
+        }
+    }
+
+    private static String sha256(byte[] bytes) {
+        try { return hex(MessageDigest.getInstance("SHA-256").digest(bytes)); }
+        catch (Exception e) { throw new IllegalStateException(e); }
+    }
+
+    private static String hex(byte[] bytes) {
+        return HexFormat.of().formatHex(bytes);
+    }
+
+    private static boolean constantEquals(String expected, String actual) {
+        return expected != null && MessageDigest.isEqual(expected.toLowerCase(Locale.ROOT).getBytes(StandardCharsets.US_ASCII),
+                actual.toLowerCase(Locale.ROOT).getBytes(StandardCharsets.US_ASCII));
+    }
+
+    private static boolean isRevision(String value) { return isBase64Sha(value); }
+    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); }
+    }
+
+    private static void deleteRecursively(Path path) throws IOException {
+        if (!Files.exists(path)) return;
+        try (var stream = Files.walk(path)) {
+            stream.sorted(Comparator.reverseOrder()).forEach(p -> {
+                try { Files.deleteIfExists(p); } catch (IOException e) { throw new UncheckedIOException(e); }
+            });
+        } catch (UncheckedIOException e) { throw e.getCause(); }
+    }
+
+    public record ApplyResult(byte code, String message) {}
+    public record Lookup(byte[] content, String revision, long generatedAt) {}
+    public record Status(String state, String revision, int accountCount, int bindingCount, long ageMillis) {}
+    private record AccountFiles(Integer accountId, byte[] v2, byte[] clash) {}
+    private record SnapshotData(Map<Integer, AccountFiles> accounts, Map<String, AccountFiles> byKeyHash, int bindingCount) {}
+    private record Snapshot(String revision, long generatedAt, Map<String, AccountFiles> byKeyHash, int accountCount, int bindingCount) {
+        boolean expired(long now, long maxAge) { return maxAge > 0 && now - generatedAt > maxAge; }
+    }
+}

src/main/java/lion/storageNode.java

@@ -1,10 +1,11 @@
 package lion;
 
-import io.netty.util.concurrent.Promise;
+import lion.Config.Config;
 import lion.Domain.GalleryTask;
 import lion.Message.*;
 import lion.Message.Main.*;
 import lion.Service.DeleteService;
+import lion.Service.SubscriptionSnapshotStore;
 import lion.Service.DownloadCheckService;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
@@ -14,7 +15,6 @@ import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import lombok.extern.slf4j.Slf4j;
 
-
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.util.HashMap;
@@ -25,8 +25,6 @@ import java.util.concurrent.locks.ReentrantLock;
 @Slf4j
 public class storageNode {
 
-    ChannelFuture channelFuture;
-
     Channel server;
 
     Channel node;
@@ -39,22 +37,28 @@ public class storageNode {
 
     ScheduledExecutorService checkThreadPool;
 
-    HashMap<Integer, Promise<AbstractMessage>> promises;
-
-    int counter;
-
     ReentrantLock lock;
 
+    final SubscriptionSnapshotStore subscriptionSnapshotStore;
+
+    final ExecutorService subscriptionApplyExecutor;
+
     public static String storagePath = "/root/gallery/gallery/";
 
-    public storageNode(){
+    public storageNode(SubscriptionSnapshotStore subscriptionSnapshotStore){
-        queue = new HashMap<>();
+        this.subscriptionSnapshotStore = subscriptionSnapshotStore;
+        this.subscriptionApplyExecutor = Executors.newSingleThreadExecutor(r -> {
+            Thread thread = new Thread(r, "subscription-snapshot-apply");
+            thread.setDaemon(true);
+            return thread;
+        });
+        queue = new ConcurrentHashMap<>();
         tempQueue = new HashMap<>();
         lock = new ReentrantLock();
-        counter = 0;
-        promises = new HashMap<>();
 
-        channelFuture = new ServerBootstrap()
+        int real_port = CustomUtil._findIdlePort(26321);
+
+        new ServerBootstrap()
                 .channel(NioServerSocketChannel.class)
                 .group(new NioEventLoopGroup())
                 .childHandler(new ChannelInitializer<NioSocketChannel>() {
@@ -64,13 +68,26 @@ public class storageNode {
                         channel.pipeline().addLast(new MessageCodec());
                         channel.pipeline().addLast(new MyChannelInboundHandlerAdapter(tempQueue));
                     }
-                })
+                }).bind(real_port);
-                .bind(26321);
+        log.info("listening on port {}", real_port);
 
-        try(Socket socket = new Socket()){
+        int i;
-            socket.connect(new InetSocketAddress("lionwebsite.xyz", 26322));
+        for(i=0; i<=20; i++) {
-        } catch (Exception ignored) {}
+            try (Socket socket = new Socket()) {
-        downloadCheckService = new DownloadCheckService(queue, promises);
+                socket.setSoTimeout(3000);
+                log.info("wake up main server on port {}", 26322 + i);
+                socket.connect(new InetSocketAddress("lionwebsite.xyz", 26322 + i));
+                byte[] bytes = socket.getInputStream().readAllBytes();
+                if(bytes.length > 0 && new String(bytes).equals("lionwebsite")) {
+                    break;
+                }
+            } catch (Exception ignored) {
+            }
+        }
+        if (i==20) {
+            log.info("server connect failed");
+        }
+        downloadCheckService = new DownloadCheckService(queue);
         checkThreadPool = Executors.newScheduledThreadPool(1);
         checkThreadPool.scheduleAtFixedRate(this::mainThread, 5, 5, TimeUnit.SECONDS);
     }
@@ -78,11 +95,14 @@ public class storageNode {
     public void mainThread(){
         try {
             lock.lock();
+            try {
                 if(!tempQueue.isEmpty()){
                     queue.putAll(tempQueue);
                     tempQueue.clear();
                 }
+            } finally {
                 lock.unlock();
+            }
             //检查，当任务状态发生变化即方法返回true时，再更新，否则return
             if (!downloadCheckService.downloadCheck()) {
                 boolean isSkip = true;
@@ -105,20 +125,23 @@ public class storageNode {
             //发送
             //上锁后再发送，避免出现发送完之后再下载完成
             lock.lock();
+            try {
                 DownloadStatusMessage downloadStatusMessage = new DownloadStatusMessage();
                 downloadStatusMessage.setGalleryTasks(queue.values().toArray(GalleryTask[]::new));
                 server.writeAndFlush(downloadStatusMessage);
 
                 queue.entrySet().removeIf(entry -> entry.getValue().is_compress_complete());
                 log.info("任务状态发送完成");
-
+            } finally {
-
                 lock.unlock();
+            }
         }catch (Exception e){
             log.error("发送任务状态时发生异常:{}", e.getMessage());
         }
     }
 
+    int counter;
+
     class MyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{
         Map<Integer, GalleryTask> queue;
 
@@ -128,6 +151,9 @@ public class storageNode {
 
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object msg) {
+            if (msg instanceof SubscriptionSnapshotMessage snapshot)
+                log.info("收到订阅快照 revision={}", shortRevision(snapshot.getRevision()));
+            else
                 log.info(String.valueOf(msg));
             AbstractMessage abstractMessage = (AbstractMessage) msg;
 
@@ -140,28 +166,55 @@ public class storageNode {
                     } else if(identityMessage.getIdentity().equals("lionwebsiteside")){
                         node = ctx.channel();
                         log.info("node上线");
-                        downloadCheckService.setNode(node);
                     }
                 }
                 case AbstractMessage.DOWNLOAD_POST_MESSAGE -> {
                     DownloadPostMessage dpm = (DownloadPostMessage) abstractMessage;
                     lock.lock();
-                    //添加到队列方法返回真说明该任务已下载完成，直接发送下载进度
+                    try {
-                    if(downloadCheckService.addToQueue(dpm.getGalleryTask())){
+                        //每次收到任务都重新检查并立即回传当前状态，供主站的单任务重试接口使用
+                        GalleryTask currentTask = downloadCheckService.addToQueue(dpm.getGalleryTask());
                         DownloadStatusMessage downloadStatusMessage = new DownloadStatusMessage();
-                        downloadStatusMessage.setGalleryTasks(new GalleryTask[]{dpm.getGalleryTask()});
+                        downloadStatusMessage.setGalleryTasks(new GalleryTask[]{currentTask});
                         server.writeAndFlush(downloadStatusMessage);
-                    }
                         log.info(String.valueOf(queue));
+                    } finally {
                         lock.unlock();
+                    }
                     ctx.writeAndFlush(new ResponseMessage(dpm.messageId, (byte) 0));
                 }
                 case AbstractMessage.DELETE_GALLERY_MESSAGE -> {
                     DeleteGalleryMessage deleteGalleryMessage = (DeleteGalleryMessage) abstractMessage;
-                    byte result = DeleteService.deleteAll(storagePath + deleteGalleryMessage.getGalleryName());
+                    String galleryName = deleteGalleryMessage.getGalleryName();
+                    lock.lock();
+                    try {
+                        // 删除任务时也要从待下载队列移除，避免继续向主站上报已经删除的任务状态。
+                        storageNode.this.queue.entrySet().removeIf(entry -> galleryName.equals(entry.getValue().getName()));
+                        queue.entrySet().removeIf(entry -> galleryName.equals(entry.getValue().getName()));
+                    } finally {
+                        lock.unlock();
+                    }
+                    byte result = DeleteService.deleteAll(storagePath + galleryName);
                     ResponseMessage responseMessage = new ResponseMessage(deleteGalleryMessage.messageId, result);
                     ctx.writeAndFlush(responseMessage);
                 }
+                case AbstractMessage.AVAILABLE_CHECK_MESSAGE -> {
+                    AvailableCheckMessage acm = (AvailableCheckMessage) abstractMessage;
+                    ResponseMessage responseMessage = new ResponseMessage(acm.messageId, (byte)0);
+                    ctx.writeAndFlush(responseMessage);
+                }
+                case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> {
+                    SubscriptionSnapshotMessage snapshotMessage = (SubscriptionSnapshotMessage) abstractMessage;
+                    if (!Config.subscriptionSyncEnabled || !ctx.channel().equals(server)) {
+                        ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, SubscriptionSnapshotStore.APPLY_INVALID));
+                        return;
+                    }
+                    subscriptionApplyExecutor.execute(() -> {
+                        SubscriptionSnapshotStore.ApplyResult result = subscriptionSnapshotStore.apply(snapshotMessage);
+                        ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, result.code()));
+                        log.info("订阅快照处理完成 revision={} result={}", shortRevision(snapshotMessage.getRevision()), result.code());
+                    });
+                }
             }
 //
 //            //修复预览
@@ -177,9 +230,11 @@ public class storageNode {
             } else if(ctx.channel().equals(node)){
                 log.info("node 下线");
                 node = null;
-                downloadCheckService.setNode(null);
             }
         }
     }
+
+    private static String shortRevision(String revision) {
+        return revision == null ? null : revision.substring(0, Math.min(12, revision.length()));
+    }
 }
-

src/main/resources/config.properties

@@ -0,0 +1,9 @@
+# 订阅快照由主站通过 Netty 长连接推送；不要在此保存上游订阅地址或 Key。
+SubscriptionSyncEnabled=false
+SubscriptionSyncSecret=
+SubscriptionDataDir=/root/gallery/storageNode/sub
+SubscriptionMaxStaleSeconds=604800
+SubscriptionMaxPayloadBytes=52428800
+SubscriptionHttpPort=8889
+SubscriptionHttpWorkers=4
+SubscriptionSocketTimeoutMs=10000

src/main/resources/logback.xml

@@ -0,0 +1,21 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <file>run.out</file>
+        <rollingPolicy class="ch.qos.logback.core.rolling.FixedWindowRollingPolicy">
+            <fileNamePattern>run.%i.out</fileNamePattern>
+            <minIndex>1</minIndex>
+            <maxIndex>10</maxIndex>
+        </rollingPolicy>
+        <triggeringPolicy class="ch.qos.logback.core.rolling.SizeBasedTriggeringPolicy">
+            <maxFileSize>10MB</maxFileSize>
+        </triggeringPolicy>
+        <encoder>
+            <pattern>%d{yyyy-MM-dd HH:mm:ss} [%level] %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="info">
+        <appender-ref ref="FILE"/>
+    </root>
+</configuration>

src/main/resources/reflect-config.json

@@ -53,6 +53,60 @@
     "allDeclaredFields" : true,
     "allPublicFields" : true
   },
+  {
+    "name": "lion.Message.Main.AvailableCheckMessage",
+    "allDeclaredConstructors" : true,
+    "allPublicConstructors" : true,
+    "allDeclaredMethods" : true,
+    "allPublicMethods" : true,
+    "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,

src/main/resources/simplelogger.properties

@@ -1,9 +0,0 @@
-
-org.slf4j.simpleLogger.defaultLogLevel=info
-
-org.slf4j.simpleLogger.showDateTime=true
-
-org.slf4j.simpleLogger.dateTimeFormat=yyyy-MM-dd HH:mm:ss
-
-org.slf4j.simpleLogger.logFile=run.out
-

src/test/java/lion/Service/DownloadCheckServiceTest.java

@@ -0,0 +1,51 @@
+package lion.Service;
+
+import lion.Domain.GalleryTask;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.file.*;
+import java.util.concurrent.ConcurrentHashMap;
+import static org.junit.jupiter.api.Assertions.*;
+
+class DownloadCheckServiceTest {
+    @Test
+    void failedCompressionRetainsSourceAndCanRetry(@TempDir Path root) throws Exception {
+        var queue = new ConcurrentHashMap<Integer, GalleryTask>();
+        var service = new DownloadCheckService(queue, false);
+        Path downloads = Files.createDirectory(root.resolve("downloads"));
+        Path source = Files.createDirectory(downloads.resolve("sample [123]"));
+        Files.writeString(source.resolve("galleryinfo.txt"), "metadata");
+        Files.writeString(source.resolve("1.jpg"), "image bytes");
+        Path storage = root.resolve("storage");
+        Files.writeString(storage, "block mkdir");
+        service.downloadPath = downloads.toString();
+        service.storagePath = storage.toString();
+        GalleryTask task = new GalleryTask();
+        task.setGid(123);
+        service.addToQueue(task);
+        service.downloadCheck();
+        service.compress();
+        assertEquals(GalleryTask.DOWNLOAD_COMPLETE, task.getStatus());
+        assertTrue(Files.exists(source.resolve("1.jpg")));
+        Files.delete(storage);
+        Files.createDirectory(storage);
+        service.addToQueue(task); // manual retry removes the backoff
+        service.downloadCheck();
+        service.compress();
+        assertEquals(GalleryTask.COMPRESS_COMPLETE, task.getStatus());
+        assertTrue(DownloadCheckService.isValidArchive(storage.resolve("sample [123]/sample [123].zip").toFile()));
+        assertFalse(Files.exists(source));
+    }
+
+    @Test
+    void corruptArchiveIsNotTreatedAsCompleted(@TempDir Path root) throws Exception {
+        Path stored = Files.createDirectories(root.resolve("stored/sample [123]"));
+        Files.writeString(stored.resolve("sample [123].zip"), "partial zip");
+        var service = new DownloadCheckService(new ConcurrentHashMap<>(), false);
+        service.storagePath = root.resolve("stored").toString();
+        service.downloadPath = root.resolve("downloads").toString();
+        GalleryTask task = new GalleryTask();
+        task.setGid(123);
+        assertFalse(service.addToQueue(task).is_compress_complete());
+    }
+}

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

@@ -0,0 +1,104 @@
+package lion.Service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lion.CustomUtil;
+import lion.Message.Main.SubscriptionAccountSnapshot;
+import lion.Message.Main.SubscriptionBindingSnapshot;
+import lion.Message.Main.SubscriptionSnapshotMessage;
+import lion.Message.Main.SubscriptionSnapshotPayload;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.ByteArrayOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.util.Base64;
+import java.util.HexFormat;
+import java.util.zip.GZIPOutputStream;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class SubscriptionSnapshotStoreTest {
+    private static final String SECRET = "snapshot-test-secret";
+    private final ObjectMapper mapper = CustomUtil.objectMapper;
+
+    @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", System.currentTimeMillis());
+
+        assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code());
+        SubscriptionSnapshotStore.Lookup v2 = store.lookup("v2", "public-key-1");
+        SubscriptionSnapshotStore.Lookup clash = store.lookup("cat", "public-key-1");
+        assertArrayEquals("v2-content".getBytes(StandardCharsets.UTF_8), v2.content());
+        assertArrayEquals("clash-content".getBytes(StandardCharsets.UTF_8), clash.content());
+        assertEquals(SubscriptionSnapshotStore.APPLY_OLD, store.apply(message).code());
+    }
+
+    @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", 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());
+        assertArrayEquals("v2-content".getBytes(StandardCharsets.UTF_8), store.lookup("v2", "public-key-1").content());
+    }
+
+    @Test
+    void loadsLastGoodSnapshotAfterRestart(@TempDir Path directory) throws Exception {
+        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);
+        second.load();
+        assertArrayEquals("clash-content".getBytes(StandardCharsets.UTF_8), second.lookup("cat", "public-key-1").content());
+    }
+
+    private SubscriptionSnapshotMessage message(String publicKey, String v2, String clash, long generatedAt) throws Exception {
+        byte[] v2Bytes = v2.getBytes(StandardCharsets.UTF_8);
+        byte[] clashBytes = clash.getBytes(StandardCharsets.UTF_8);
+        SubscriptionAccountSnapshot account = new SubscriptionAccountSnapshot();
+        account.setAccountId(1);
+        account.setEnabled(true);
+        account.setV2ContentBase64(Base64.getEncoder().encodeToString(v2Bytes));
+        account.setV2Sha256(sha256(v2Bytes));
+        account.setClashContentBase64(Base64.getEncoder().encodeToString(clashBytes));
+        account.setClashSha256(sha256(clashBytes));
+        SubscriptionBindingSnapshot binding = new SubscriptionBindingSnapshot();
+        binding.setPublicKeySha256(sha256(publicKey.getBytes(StandardCharsets.UTF_8)));
+        binding.setAccountId(1);
+        SubscriptionSnapshotPayload payload = new SubscriptionSnapshotPayload();
+        payload.setSchemaVersion(1);
+        payload.setAccounts(java.util.List.of(account));
+        payload.setBindings(java.util.List.of(binding));
+        byte[] json = mapper.writeValueAsBytes(payload);
+        byte[] compressed = gzip(json);
+        SubscriptionSnapshotMessage message = new SubscriptionSnapshotMessage();
+        message.setSchemaVersion(1);
+        message.setRevision(sha256(json));
+        message.setGeneratedAt(generatedAt);
+        message.setPayloadBase64(Base64.getEncoder().encodeToString(compressed));
+        message.setPayloadSha256(sha256(compressed));
+        String input = "1\n" + message.getRevision() + "\n" + generatedAt + "\n" + message.getPayloadSha256();
+        Mac mac = Mac.getInstance("HmacSHA256");
+        mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+        message.setSignature(hex(mac.doFinal(input.getBytes(StandardCharsets.UTF_8))));
+        return message;
+    }
+
+    private static byte[] gzip(byte[] bytes) throws Exception {
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+        try (GZIPOutputStream gzip = new GZIPOutputStream(output)) { gzip.write(bytes); }
+        return output.toByteArray();
+    }
+
+    private static String sha256(byte[] bytes) throws Exception {
+        return hex(MessageDigest.getInstance("SHA-256").digest(bytes));
+    }
+
+    private static String hex(byte[] bytes) { return HexFormat.of().formatHex(bytes); }
+}