diff --git a/src/main/java/lion/CustomUtil.java b/src/main/java/lion/CustomUtil.java index db3fc86..e65b940 100644 --- a/src/main/java/lion/CustomUtil.java +++ b/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(); } } -} \ No newline at end of file +} diff --git a/src/main/java/lion/MultiThreadedHTTPServer.java b/src/main/java/lion/MultiThreadedHTTPServer.java index 9419e8a..e5ea61a 100644 --- a/src/main/java/lion/MultiThreadedHTTPServer.java +++ b/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); diff --git a/src/main/java/lion/Service/DownloadCheckService.java b/src/main/java/lion/Service/DownloadCheckService.java index 0c77d88..39d6711 100644 --- a/src/main/java/lion/Service/DownloadCheckService.java +++ b/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 retryAfter = new ConcurrentHashMap<>(); + /** + * 已校验通过的归档缓存：路径 → (归档大小, 校验时刻)。 + * + *

归档一旦落盘就不再变化（发布用 ATOMIC_MOVE），因此同一路径只有在大小变化时 + * 才需要重新校验。没有它时，每次 {@link #addToQueue} 都会把候选 ZIP 整包读取并 + * 重算 CRC——主站重连触发 resetUndone 时会对所有未完成任务走一遍，纯属浪费磁盘 I/O。 + */ + final Map verifiedArchives = new ConcurrentHashMap<>(); + + private static final long ARCHIVE_REVERIFY_MILLIS = 3_600_000L; + + private record VerifiedArchive(long size, long verifiedAt) {} + + /** + * 完成通知的专用线程。 + * + *

{@link CustomUtil#notifyMe} 是同步 HTTP，原先在 {@link #addToQueue} 里直接调用， + * 而 addToQueue 由 Netty IO 线程持节点主锁执行，一次慢请求就会把整个节点的状态上报 + * 与删除处理一起卡住。改为异步投递后，外呼延迟不再占用锁。 + */ + private final ExecutorService notificationExecutor; + + /** 是否发送完成通知；仅生产构造（启动调度器）时启用，单元测试保持静默不外呼。 */ + private final boolean notificationsEnabled; + public DownloadCheckService(Map queue){ this(queue, true); } @@ -38,6 +65,12 @@ public class DownloadCheckService { DownloadCheckService(Map 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) diff --git a/src/main/java/lion/Service/SubscriptionSnapshotStore.java b/src/main/java/lion/Service/SubscriptionSnapshotStore.java index 2d24a2b..b84c813 100644 --- a/src/main/java/lion/Service/SubscriptionSnapshotStore.java +++ b/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 current = new AtomicReference<>(); + /** + * 最近一次收到并校验通过的主站快照的时刻（本地接收时间）。 + * + *

过期判定必须基于它而不是内容里的 {@code generatedAt}：revision 是内容寻址的， + * 内容不变时主站每分钟重发同一 revision、{@code generatedAt} 却一直在刷新， + * 节点按 APPLY_OLD 丢弃后新鲜度永远停在首次接收那天——于是主站明明在线且持续同步， + * 备机也会在第 7 天被判过期并开始返回 503。改用本地接收时间后，「过期」才真正 + * 表示「主站已失联」，与设计文档一致。 + * + *

接收时间取自节点本地时钟、不参与签名，因此攻击者无法通过重放旧 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 accounts, Map byKeyHash, int bindingCount) {} - private record Snapshot(String revision, long generatedAt, Map 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 byKeyHash, int accountCount, int bindingCount) {} } diff --git a/src/main/java/lion/storageNode.java b/src/main/java/lion/storageNode.java index 86eb467..d812adf 100644 --- a/src/main/java/lion/storageNode.java +++ b/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(); diff --git a/src/test/java/lion/MultiThreadedHTTPServerTest.java b/src/test/java/lion/MultiThreadedHTTPServerTest.java new file mode 100644 index 0000000..5faa3c5 --- /dev/null +++ b/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 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 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 params = MultiThreadedHTTPServer.parseRequestLine( + "/download?AuthCode=alone&token=YWJjZA=="); + + assertEquals("YWJjZA==", params.get("token"), "值内的 '=' 不得被截断"); + assertEquals("alone", params.get("AuthCode")); + } + + /** URL 编码的值必须解码回原文。 */ + @Test + void decodesPercentEncodedValues() { + Map params = MultiThreadedHTTPServer.parseRequestLine( + "/download?AuthCode=alone&name=a%20b"); + assertEquals("a b", params.get("name")); + } + + /** null 请求行返回 null，交由调用方兜底。 */ + @Test + void returnsNullForNullRequestLine() { + assertNull(MultiThreadedHTTPServer.parseRequestLine(null)); + } +} diff --git a/src/test/java/lion/Service/DownloadCheckServiceTest.java b/src/test/java/lion/Service/DownloadCheckServiceTest.java index 6b5c3f3..dbe24da 100644 --- a/src/test/java/lion/Service/DownloadCheckServiceTest.java +++ b/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(), "第二次应命中校验缓存且结论一致"); + } } diff --git a/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java b/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java index 006cc0d..1e515c0 100644 --- a/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java +++ b/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);