修复订阅快照过期误判，并加固下载解析与完成通知

过期判定（真实缺陷）：
- 节点原先按内容里的 generatedAt 判过期，而 revision 是内容寻址的。内容长期不变时
  主站每分钟重发同一 revision 会被判为 APPLY_OLD 直接丢弃，新鲜度永远停在首次接收
  那天——于是主站明明在线且持续同步，备机仍在第 7 天开始返回 503。
- 改为按「最近一次成功接收并校验通过的主站快照时刻」判定：同 revision 重发会刷新
  新鲜度；重启时以落盘内容的生成时间作为起点，避免重启即续期。
- 该时刻取自节点本地时钟、不参与签名，重放旧 revision 无法续期。

下载 HTTP 服务：
- 请求行改按 ISO-8859-1 显式解码（原依赖平台默认字符集）。
- 查询串与键值拆分的 split 加 limit=2：参数值含 '='（如 base64 padding）不再被截断。
- sendFileRange 缓冲 1 KiB -> 64 KiB，降低大包下载的系统调用开销。

队列上报与通知：
- server/node 通道加 volatile；主站未连接或通道失效时跳过状态上报，不再以 NPE 形式
  被外层 catch 吞掉后每 5 秒刷一条 error（队列保留，等重连重放）。
- 无待上报任务时不再发送空数组。
- 已下载完成的通知移出 addToQueue 的持锁路径并改为异步，避免同步 HTTP 卡住节点主锁；
  且仅生产构造启用，测试不外呼。

已下载完成的重复校验：
- 归档校验结果按「路径 + 大小」缓存（大小变化或超一小时才重算），避免重连时把所有
  未完成任务整包读取重算 CRC。

测试：13 项全过（新增 MultiThreadedHTTPServerTest 5 项；SubscriptionSnapshotStoreTest
增加「同 revision 重发续期」「超期未收到推送才过期」；DownloadCheckServiceTest 增加
有效归档跨重复入队判定一致）。

src/main/java/lion/CustomUtil.java

@@ -79,7 +79,8 @@ public class CustomUtil {
 
         try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r")) {
             randomAccessFile.seek(startByte);
-            byte[] buffer = new byte[1024];
+            // 1 KiB 每次系统调用太小，大包下载时开销明显；64 KiB 是常见的折中值。
+            byte[] buffer = new byte[64 * 1024];
             int bytesRead;
             long bytesRemaining = endByte - startByte + 1;
             while (bytesRemaining > 0 && (bytesRead = randomAccessFile.read(buffer, 0, (int) Math.min(buffer.length, bytesRemaining))) != -1) {
@@ -91,4 +92,4 @@ public class CustomUtil {
             responseStream.close();
         }
     }
-}
+}

src/main/java/lion/MultiThreadedHTTPServer.java

@@ -44,7 +44,8 @@ public class MultiThreadedHTTPServer {
     private static void handleClientRequest(Socket clientSocket) {
         String fileName = "";
         try {
-            BufferedReader requestReader = new BufferedReader(new InputStreamReader(clientSocket.getInputStream()));
+            BufferedReader requestReader = new BufferedReader(
+                    new InputStreamReader(clientSocket.getInputStream(), StandardCharsets.ISO_8859_1));
             String requestLine = requestReader.readLine();
 
             if (requestLine == null) {
@@ -154,7 +155,7 @@ public class MultiThreadedHTTPServer {
             return null;
 
         if (requestLine.contains("?")) {
-            String[] requestParts = requestLine.split("\\?");
+            String[] requestParts = requestLine.split("\\?", 2);
             String path = requestParts[0];
             queryParams.put("path", path);
 
@@ -164,7 +165,8 @@ public class MultiThreadedHTTPServer {
 
             // Split the query string into individual parameter key-value pairs
             for (String paramPair : paramPairs) {
-                String[] keyValue = paramPair.split("=");
+                // limit=2：参数值里含有 '='（例如 base64 结尾的 padding）时不能被截断。
+                String[] keyValue = paramPair.split("=", 2);
                 if (keyValue.length == 2) {
                     String key = URLDecoder.decode(keyValue[0], StandardCharsets.UTF_8);
                     String value = URLDecoder.decode(keyValue[1], StandardCharsets.UTF_8);

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

@@ -8,6 +8,8 @@ import lombok.extern.slf4j.Slf4j;
 
 import java.io.*;
 import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.nio.file.*;
@@ -30,6 +32,31 @@ public class DownloadCheckService {
 
     final Map<Integer, Long> retryAfter = new ConcurrentHashMap<>();
 
+    /**
+     * 已校验通过的归档缓存：路径 → (归档大小, 校验时刻)。
+     *
+     * <p>归档一旦落盘就不再变化（发布用 ATOMIC_MOVE），因此同一路径只有在大小变化时
+     * 才需要重新校验。没有它时，每次 {@link #addToQueue} 都会把候选 ZIP 整包读取并
+     * 重算 CRC——主站重连触发 resetUndone 时会对所有未完成任务走一遍，纯属浪费磁盘 I/O。
+     */
+    final Map<String, VerifiedArchive> verifiedArchives = new ConcurrentHashMap<>();
+
+    private static final long ARCHIVE_REVERIFY_MILLIS = 3_600_000L;
+
+    private record VerifiedArchive(long size, long verifiedAt) {}
+
+    /**
+     * 完成通知的专用线程。
+     *
+     * <p>{@link CustomUtil#notifyMe} 是同步 HTTP，原先在 {@link #addToQueue} 里直接调用，
+     * 而 addToQueue 由 Netty IO 线程持节点主锁执行，一次慢请求就会把整个节点的状态上报
+     * 与删除处理一起卡住。改为异步投递后，外呼延迟不再占用锁。
+     */
+    private final ExecutorService notificationExecutor;
+
+    /** 是否发送完成通知；仅生产构造（启动调度器）时启用，单元测试保持静默不外呼。 */
+    private final boolean notificationsEnabled;
+
     public DownloadCheckService(Map<Integer, GalleryTask> queue){
         this(queue, true);
     }
@@ -38,6 +65,12 @@ public class DownloadCheckService {
     DownloadCheckService(Map<Integer, GalleryTask> queue, boolean startScheduler){
         this.queue = queue;
         compress_queue = new ArrayList<>(0);
+        this.notificationsEnabled = startScheduler;
+        this.notificationExecutor = Executors.newSingleThreadExecutor(r -> {
+            Thread thread = new Thread(r, "storage-node-notify");
+            thread.setDaemon(true);
+            return thread;
+        });
         if (startScheduler) {
             convert_thread = new ScheduledThreadPoolExecutor(1);
             convert_thread.scheduleWithFixedDelay(this::compress, 0, 5, TimeUnit.SECONDS);
@@ -199,7 +232,7 @@ public class DownloadCheckService {
             queue.remove(galleryTask.getGid());
             galleryTask.setName(storedDirectory.getName());
             galleryTask.setStatus(GalleryTask.COMPRESS_COMPLETE);
-            CustomUtil.notifyMe(String.format("任务:%s在添加时已下载完成，更新任务状态", galleryTask.getName()));
+            notifyAlreadyStored(galleryTask.getName());
             return galleryTask;
         }
 
@@ -242,12 +275,39 @@ public class DownloadCheckService {
 
         for(File directory : directories){
             if(matchesGid(directory.getName(), gid)
-                    && isValidArchive(new File(directory, directory.getName() + ".zip")))
+                    && isVerifiedArchive(new File(directory, directory.getName() + ".zip")))
                 return directory;
         }
         return null;
     }
 
+    /**
+     * 压缩包校验带缓存：归档按大小不变，校验结果可复用。
+     * 大小变化（重新压缩）或距上次校验超过一小时才重算，避免重复整包读取。
+     */
+    private boolean isVerifiedArchive(File archive){
+        if(!archive.isFile())
+            return false;
+        long size = archive.length();
+        long now = System.currentTimeMillis();
+        VerifiedArchive cached = verifiedArchives.get(archive.getPath());
+        if(cached != null && cached.size() == size && now - cached.verifiedAt() < ARCHIVE_REVERIFY_MILLIS)
+            return true;
+        if(!isValidArchive(archive)){
+            verifiedArchives.remove(archive.getPath());
+            return false;
+        }
+        verifiedArchives.put(archive.getPath(), new VerifiedArchive(size, now));
+        return true;
+    }
+
+    private void notifyAlreadyStored(String taskName){
+        if(!notificationsEnabled)
+            return;
+        notificationExecutor.execute(() ->
+                CustomUtil.notifyMe(String.format("任务:%s在添加时已下载完成，更新任务状态", taskName)));
+    }
+
     private File findDirectoryByGid(File parentDirectory, int gid){
         File[] directories = parentDirectory.listFiles(File::isDirectory);
         if(directories == null)

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

@@ -38,6 +38,18 @@ public final class SubscriptionSnapshotStore {
     private final int maxPayloadBytes;
     private final ObjectMapper objectMapper = CustomUtil.objectMapper;
     private final AtomicReference<Snapshot> current = new AtomicReference<>();
+    /**
+     * 最近一次收到并校验通过的主站快照的时刻（本地接收时间）。
+     *
+     * <p>过期判定必须基于它而不是内容里的 {@code generatedAt}：revision 是内容寻址的，
+     * 内容不变时主站每分钟重发同一 revision、{@code generatedAt} 却一直在刷新，
+     * 节点按 APPLY_OLD 丢弃后新鲜度永远停在首次接收那天——于是主站明明在线且持续同步，
+     * 备机也会在第 7 天被判过期并开始返回 503。改用本地接收时间后，「过期」才真正
+     * 表示「主站已失联」，与设计文档一致。
+     *
+     * <p>接收时间取自节点本地时钟、不参与签名，因此攻击者无法通过重放旧 revision 续期。
+     */
+    private volatile long lastSyncAt;
 
     public SubscriptionSnapshotStore(Path root, String syncSecret, long maxStaleSeconds, int maxPayloadBytes) {
         this.root = Objects.requireNonNull(root);
@@ -72,6 +84,9 @@ public final class SubscriptionSnapshotStore {
             try {
                 Snapshot snapshot = loadSnapshot(root.resolve("snapshots").resolve(revision));
                 current.set(snapshot);
+                // 重启后还没有收到过主站快照，先以落盘内容的生成时间作为新鲜度起点，
+                // 否则重启即视为「刚同步过」，会让超期快照被错误续期。
+                lastSyncAt = snapshot.generatedAt();
                 writePointer(revision);
                 log.info("加载订阅快照成功 revision={} accounts={} bindings={}", shortRevision(revision), snapshot.accountCount(), snapshot.bindingCount());
                 return;
@@ -106,8 +121,11 @@ public final class SubscriptionSnapshotStore {
             SnapshotData data = validatePayload(payload);
             Snapshot old = current.get();
             if (old != null) {
-                if (message.getRevision().equals(old.revision()))
+                if (message.getRevision().equals(old.revision())) {
+                    // 内容未变但主站仍在同步：刷新新鲜度，避免备机因「内容长期不变」而误判过期。
+                    lastSyncAt = System.currentTimeMillis();
                     return new ApplyResult(APPLY_OLD, "revision 已存在");
+                }
                 if (message.getGeneratedAt() < old.generatedAt())
                     return new ApplyResult(APPLY_OLD, "快照时间早于当前版本");
             }
@@ -139,6 +157,7 @@ public final class SubscriptionSnapshotStore {
             Snapshot snapshot = new Snapshot(message.getRevision(), message.getGeneratedAt(), data.byKeyHash,
                     data.accounts.size(), data.bindingCount);
             current.set(snapshot);
+            lastSyncAt = System.currentTimeMillis();
             cleanupOldSnapshots(message.getRevision());
             return new ApplyResult(APPLY_SUCCESS, "同步成功");
         } catch (Exception e) {
@@ -162,7 +181,7 @@ public final class SubscriptionSnapshotStore {
 
     public Lookup lookup(String client, String publicKey) {
         Snapshot snapshot = current.get();
-        if (snapshot == null || snapshot.expired(System.currentTimeMillis(), maxStaleMillis))
+        if (snapshot == null || isStale(System.currentTimeMillis()))
             return null;
         AccountFiles account = snapshot.byKeyHash().get(sha256(publicKey.getBytes(StandardCharsets.UTF_8)));
         if (account == null)
@@ -174,9 +193,14 @@ public final class SubscriptionSnapshotStore {
         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);
+        long now = System.currentTimeMillis();
+        long age = Math.max(0, now - lastSyncAt);
+        return new Status(isStale(now) ? "expired" : "ready", snapshot.revision(), snapshot.accountCount(), snapshot.bindingCount(), age);
+    }
+
+    /** 距最近一次成功接收主站快照是否已超过最大有效期。 */
+    private boolean isStale(long now) {
+        return maxStaleMillis > 0 && now - lastSyncAt > maxStaleMillis;
     }
 
     private Snapshot loadSnapshot(Path directory) throws IOException {
@@ -323,7 +347,5 @@ public final class SubscriptionSnapshotStore {
     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; }
-    }
+    private record Snapshot(String revision, long generatedAt, Map<String, AccountFiles> byKeyHash, int accountCount, int bindingCount) {}
 }

src/main/java/lion/storageNode.java

@@ -25,9 +25,10 @@ import java.util.concurrent.locks.ReentrantLock;
 @Slf4j
 public class storageNode {
 
-    Channel server;
+    // 分别由 Netty IO 线程写入、由 5 秒定时线程读取，必须保证可见性。
+    volatile Channel server;
 
-    Channel node;
+    volatile Channel node;
 
     DownloadCheckService downloadCheckService;
 
@@ -122,13 +123,21 @@ public class storageNode {
                     return;
                 }
             }
+            //主站未连接时不上报。原先直接 server.writeAndFlush 会在 server 为 null 时抛 NPE，
+            //被外层 catch 吞掉后每 5 秒刷一条 error；队列保留，等主站重连时重放。
+            Channel target = server;
+            if (target == null || !target.isActive())
+                return;
+            //没有待上报任务时不必发送空数组（此时 downloadCheck 为 false 且无压缩完成任务）。
+            if (queue.isEmpty())
+                return;
             //发送
             //上锁后再发送，避免出现发送完之后再下载完成
             lock.lock();
             try {
                 DownloadStatusMessage downloadStatusMessage = new DownloadStatusMessage();
                 downloadStatusMessage.setGalleryTasks(queue.values().toArray(GalleryTask[]::new));
-                server.writeAndFlush(downloadStatusMessage);
+                target.writeAndFlush(downloadStatusMessage);
 
                 queue.entrySet().removeIf(entry -> entry.getValue().is_compress_complete());
                 log.info("任务状态发送完成");
@@ -176,7 +185,9 @@ public class storageNode {
                         GalleryTask currentTask = downloadCheckService.addToQueue(dpm.getGalleryTask());
                         DownloadStatusMessage downloadStatusMessage = new DownloadStatusMessage();
                         downloadStatusMessage.setGalleryTasks(new GalleryTask[]{currentTask});
-                        server.writeAndFlush(downloadStatusMessage);
+                        Channel target = server;
+                        if (target != null && target.isActive())
+                            target.writeAndFlush(downloadStatusMessage);
                         log.info(String.valueOf(queue));
                     } finally {
                         lock.unlock();

src/test/java/lion/MultiThreadedHTTPServerTest.java

@@ -0,0 +1,61 @@
+package lion;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * 下载 HTTP 服务的请求行解析。
+ * 这条链路只接受主站 IP 的请求，解析细节直接决定能否按 gid 找到正确的压缩包。
+ */
+class MultiThreadedHTTPServerTest {
+
+    /** 普通请求：path 与各查询参数都要被拆出来。 */
+    @Test
+    void parsesPathAndQueryParameters() {
+        Map<String, String> params = MultiThreadedHTTPServer.parseRequestLine(
+                "/download?AuthCode=abc&gid=12345");
+
+        assertEquals("/download", params.get("path"));
+        assertEquals("abc", params.get("AuthCode"));
+        assertEquals("12345", params.get("gid"));
+    }
+
+    /** 无查询串时只回 path，且不应抛异常。 */
+    @Test
+    void parsesPathWithoutQueryString() {
+        Map<String, String> params = MultiThreadedHTTPServer.parseRequestLine("/archive/file.zip");
+        assertEquals("/archive/file.zip", params.get("path"));
+        assertNull(params.get("gid"));
+    }
+
+    /**
+     * 参数值里含 '=' 时必须完整保留。
+     * 回归：split("=") 无 limit 会在第一个 '=' 处截断，
+     * base64 padding（结尾的 '='）等取值会被悄悄改短。
+     */
+    @Test
+    void keepsEqualsSignInsideParameterValue() {
+        Map<String, String> params = MultiThreadedHTTPServer.parseRequestLine(
+                "/download?AuthCode=alone&token=YWJjZA==");
+
+        assertEquals("YWJjZA==", params.get("token"), "值内的 '=' 不得被截断");
+        assertEquals("alone", params.get("AuthCode"));
+    }
+
+    /** URL 编码的值必须解码回原文。 */
+    @Test
+    void decodesPercentEncodedValues() {
+        Map<String, String> params = MultiThreadedHTTPServer.parseRequestLine(
+                "/download?AuthCode=alone&name=a%20b");
+        assertEquals("a b", params.get("name"));
+    }
+
+    /** null 请求行返回 null，交由调用方兜底。 */
+    @Test
+    void returnsNullForNullRequestLine() {
+        assertNull(MultiThreadedHTTPServer.parseRequestLine(null));
+    }
+}

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

@@ -48,4 +48,31 @@ class DownloadCheckServiceTest {
         task.setGid(123);
         assertFalse(service.addToQueue(task).is_compress_complete());
     }
+
+    /**
+     * 已有有效归档时必须判为完成，且重复添加（命中校验缓存）行为保持一致。
+     * 校验缓存按「路径 + 大小」复用，目的是避免每次 addToQueue 都把整包读一遍，
+     * 但它绝不能让有效归档被误判为未完成。
+     */
+    @Test
+    void validArchiveIsRecognizedAcrossRepeatedQueueAdds(@TempDir Path root) throws Exception {
+        Path stored = Files.createDirectories(root.resolve("stored/sample [123]"));
+        Path zip = stored.resolve("sample [123].zip");
+        try (var out = new java.util.zip.ZipOutputStream(Files.newOutputStream(zip))) {
+            out.putNextEntry(new java.util.zip.ZipEntry("1.jpg"));
+            out.write("image bytes".getBytes(java.nio.charset.StandardCharsets.UTF_8));
+            out.closeEntry();
+        }
+        var service = new DownloadCheckService(new ConcurrentHashMap<>(), false);
+        service.storagePath = root.resolve("stored").toString();
+        service.downloadPath = root.resolve("downloads").toString();
+
+        GalleryTask first = new GalleryTask();
+        first.setGid(123);
+        assertTrue(service.addToQueue(first).is_compress_complete(), "存在有效归档时应判为完成");
+
+        GalleryTask second = new GalleryTask();
+        second.setGid(123);
+        assertTrue(service.addToQueue(second).is_compress_complete(), "第二次应命中校验缓存且结论一致");
+    }
 }

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

@@ -58,6 +58,46 @@ class SubscriptionSnapshotStoreTest {
         assertArrayEquals("clash-content".getBytes(StandardCharsets.UTF_8), second.lookup("cat", "public-key-1").content());
     }
 
+    /**
+     * 回归：节点曾按「内容里的 generatedAt」判过期，而 revision 是内容寻址的——
+     * 内容长期不变时主站每分钟重发同一 revision 会被判为 APPLY_OLD 直接丢弃，
+     * 新鲜度永远停在首次接收那天，于是主站在线且持续同步，备机仍在第 7 天开始 503。
+     * 现在过期只取决于「最近一次成功接收主站快照的时刻」。
+     */
+    @Test
+    void repeatedSameRevisionRefreshesFreshnessAndNeverExpires(@TempDir Path directory) throws Exception {
+        // 有效期 1 秒，便于在用例内观察「不续期会过期、续期后恢复」。
+        SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 1, 1024 * 1024);
+        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis());
+
+        assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code());
+        assertEquals("ready", store.status().state());
+
+        // 超过有效期且期间没有任何推送 → 过期（证明判定确实生效，不是恒 ready）
+        Thread.sleep(1_200);
+        assertEquals("expired", store.status().state(), "超过有效期且无推送应判过期");
+
+        // 同 revision 重发：仍是 APPLY_OLD（内容幂等），但必须刷新新鲜度
+        assertEquals(SubscriptionSnapshotStore.APPLY_OLD, store.apply(message).code());
+        assertEquals("ready", store.status().state(), "重新收到同 revision 后应恢复可用");
+        assertNotNull(store.lookup("v2", "public-key-1"), "持续同步期间必须能取到订阅");
+    }
+
+    /** 超过有效期后仍未收到任何快照，才判过期并停止分发。 */
+    @Test
+    void expiresOnlyAfterNoSnapshotArrivesWithinMaxAge(@TempDir Path directory) throws Exception {
+        SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
+        // 内容生成于 2 小时前，且重启后再没收到主站推送 → 超期。
+        SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content",
+                System.currentTimeMillis() - 7_200_000L);
+        assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code());
+
+        SubscriptionSnapshotStore reloaded = new SubscriptionSnapshotStore(directory, SECRET, 3600, 1024 * 1024);
+        reloaded.load();
+        assertEquals("expired", reloaded.status().state(), "重启后未再收到快照，超期应判过期");
+        assertNull(reloaded.lookup("v2", "public-key-1"), "过期快照不得继续分发");
+    }
+
     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);