默认独立于订阅数据目录：任务日志与订阅快照的生命周期、备份策略和损坏影响面都不同， + * 混在一处会让其中一方的清理或迁移误伤另一方。 + */ + public static String taskJournalDir = "/root/gallery/storageNode/task-journal"; + public static void loadConfig(){ loadConfig(CONFIG_PATH); } @@ -72,6 +80,7 @@ public class Config { adminDownloadCode = System.getenv().getOrDefault("STORAGE_DOWNLOAD_ADMIN_CODE", value(prop, "AdminDownloadCode", "")); adminDownloadRoot = value(prop, "AdminDownloadRoot", "/root/gallery/gallery"); + taskJournalDir = value(prop, "TaskJournalDir", "/root/gallery/storageNode/task-journal"); if (subscriptionSyncEnabled && subscriptionSyncSecret.isBlank()) throw new IllegalStateException("启用订阅同步时必须配置 SUBSCRIPTION_SYNC_SECRET"); } diff --git a/src/main/java/lion/CustomUtil.java b/src/main/java/lion/CustomUtil.java index da629cb..88f6e20 100644 --- a/src/main/java/lion/CustomUtil.java +++ b/src/main/java/lion/CustomUtil.java @@ -5,13 +5,18 @@ import cn.hutool.http.HttpResponse; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.extern.slf4j.Slf4j; -import java.io.*; +import java.io.IOException; import java.net.ServerSocket; -import java.net.Socket; -import java.net.SocketException; import java.net.URLEncoder; import java.nio.charset.StandardCharsets; +/** + * 通用工具：外部通知、空闲端口探测，以及共享的 {@link ObjectMapper}。 + * + *
原先手写的 HTTP 响应工具（{@code getRequestHeader} / {@code sendErrorResponse} / + * {@code sendFileRange}）已随 8888 文件服务迁移到 JDK {@code HttpServer} 而移除； + * 文件与 Range 的语义现在由 {@code lion.Http.FileResponseFactory} 统一负责。 + */ @Slf4j public class CustomUtil { @@ -40,74 +45,4 @@ public class CustomUtil { } 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("
外壳已从手工 {@code ServerSocket} 改为与下载服务相同的 JDK + * {@code com.sun.net.httpserver.HttpServer}（{@link HttpFileServer}）。路径与语义保持不变： + * + *
{@link #run()} 保留为旧入口（{@code Main} 用它起线程），内部改为阻塞等待，直到线程被中断。
+ */
@Slf4j
public final class BackupSubServer implements Runnable {
+
+ /** 公开 Key 的合法长度下限，过短直接按不存在处理，避免无谓查表。 */
+ private static final int MIN_KEY_LENGTH = 6;
+ private static final int MAX_KEY_LENGTH = 512;
+
+ private static final String V2_CLIENT = "v2";
+ private static final String CLASH_CLIENT = "cat";
+
private final SubscriptionSnapshotStore snapshotStore;
private final int port;
- private final ExecutorService workers;
+ private final int workerCount;
+ private volatile HttpFileServer server;
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());
+ this.workerCount = Math.max(1, workerCount);
}
+ /** 旧入口：绑定端口、开始接收请求，并阻塞到线程被中断。 */
@Override
public void run() {
- try (ServerSocket serverSocket = new ServerSocket(port)) {
- log.info("备机订阅服务监听端口 {}", port);
+ try {
+ HttpFileServer running = startServer();
+ server = running;
+ log.info("备机订阅服务监听端口 {}", running.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) { }
+ Thread.sleep(60_000);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
}
}
+ running.close();
} 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 只处理单区间（与旧实现的 {@code split(",", 2)[0]} 行为一致），但补齐旧实现缺失的
+ * 后缀区间 {@code bytes=-n} 与开放结尾 {@code bytes=a-}，并把「无法满足」与「语法非法」
+ * 分成两种结果：前者回 416，后者按 RFC 忽略 Range 返回完整实体。
+ */
+public final class ByteRange {
+
+ /** 解析结论。 */
+ public enum Kind {
+ /** 忽略 Range（没有该头、语法非法或不受支持），按完整实体返回 200。 */
+ FULL,
+ /** 返回 206 与指定区间。 */
+ PARTIAL,
+ /** 语法合法但落在实体之外，返回 416。 */
+ UNSATISFIABLE
+ }
+
+ /** 一次 Range 解析的结果；{@code start}/{@code endInclusive} 仅在 {@link Kind#PARTIAL} 时有意义。 */
+ public record Resolution(Kind kind, long start, long endInclusive) {
+ public boolean partial() {
+ return kind == Kind.PARTIAL;
+ }
+
+ public boolean unsatisfiable() {
+ return kind == Kind.UNSATISFIABLE;
+ }
+
+ /** 区间字节数。 */
+ public long length() {
+ return endInclusive - start + 1;
+ }
+
+ /** 形如 {@code bytes 4-9/10} 的响应头值。 */
+ public String contentRange(long totalLength) {
+ return "bytes " + start + "-" + endInclusive + "/" + totalLength;
+ }
+ }
+
+ private static final Resolution FULL = new Resolution(Kind.FULL, 0, -1);
+ private static final Resolution UNSATISFIABLE = new Resolution(Kind.UNSATISFIABLE, 0, -1);
+
+ private ByteRange() {
+ }
+
+ /**
+ * 解析 Range 头。
+ *
+ * @param headerValue 原始 {@code Range} 头，可为 null
+ * @param fileLength 实体总长度
+ */
+ public static Resolution parse(String headerValue, long fileLength) {
+ if (headerValue == null)
+ return FULL;
+ String value = headerValue.trim();
+ if (!value.toLowerCase(Locale.ROOT).startsWith("bytes="))
+ return FULL;
+ value = value.substring("bytes=".length()).trim();
+ // 多区间只取第一段，与旧实现一致。
+ int comma = value.indexOf(',');
+ if (comma >= 0)
+ value = value.substring(0, comma).trim();
+ if (value.isEmpty())
+ return FULL;
+
+ int dash = value.indexOf('-');
+ if (dash < 0)
+ return FULL;
+ String first = value.substring(0, dash).trim();
+ String second = value.substring(dash + 1).trim();
+
+ if (first.isEmpty()) {
+ // 后缀区间：bytes=-n 取最后 n 字节。
+ long suffix = parseLong(second);
+ if (suffix < 0)
+ return FULL;
+ if (suffix == 0 || fileLength == 0)
+ return UNSATISFIABLE;
+ return partial(Math.max(0, fileLength - suffix), fileLength - 1);
+ }
+
+ long start = parseLong(first);
+ if (start < 0)
+ return FULL;
+ if (start >= fileLength)
+ return UNSATISFIABLE;
+
+ long end;
+ if (second.isEmpty()) {
+ end = fileLength - 1;
+ } else {
+ end = parseLong(second);
+ if (end < 0)
+ return FULL;
+ if (end < start)
+ return UNSATISFIABLE;
+ end = Math.min(end, fileLength - 1);
+ }
+ return partial(start, end);
+ }
+
+ private static Resolution partial(long start, long endInclusive) {
+ return new Resolution(Kind.PARTIAL, start, endInclusive);
+ }
+
+ /** 非数字、空串与溢出一律按「无法解析」处理，由调用方退回完整实体。 */
+ private static long parseLong(String value) {
+ if (value == null || value.isEmpty())
+ return -1;
+ try {
+ return Long.parseLong(value);
+ } catch (NumberFormatException notANumber) {
+ return -1;
+ }
+ }
+}
diff --git a/src/main/java/lion/Http/FileResponseFactory.java b/src/main/java/lion/Http/FileResponseFactory.java
new file mode 100644
index 0000000..2aec6ce
--- /dev/null
+++ b/src/main/java/lion/Http/FileResponseFactory.java
@@ -0,0 +1,82 @@
+package lion.Http;
+
+import java.io.File;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * 把「文件 + Range 头」变成响应模型，统一 200/206/416 的语义。
+ *
+ * 下载（8888）与订阅（8889）两条链路共用，避免又出现两套手写 Range 逻辑。
+ */
+public final class FileResponseFactory {
+
+ /** 不能作为 Content-Disposition 文件名的字符（引号、反斜杠与控制字符）统一替换。 */
+ private static final String UNSAFE_FILENAME_CHARS = "[\r\n\"\\\\]";
+
+ private FileResponseFactory() {
+ }
+
+ /** 无 Content-Disposition 的文件响应。 */
+ public static HttpResponseModel forFile(String rangeHeader, File file, String contentDispositionFileName) {
+ return forFile(rangeHeader, file, contentDispositionFileName, "application/octet-stream");
+ }
+
+ /**
+ * 内存正文的区间响应（订阅内容走这里）。
+ *
+ * 与 {@link #forFile} 共用 {@link ByteRange} 的 200/206/416 判定，避免两套实现漂移。
+ */
+ public static HttpResponseModel forBytes(String rangeHeader, byte[] content, String contentType,
+ String contentDispositionFileName) {
+ byte[] body = content == null ? new byte[0] : content;
+ String disposition = contentDispositionFileName == null
+ ? null : contentDisposition(contentDispositionFileName);
+
+ ByteRange.Resolution range = ByteRange.parse(rangeHeader, body.length);
+ if (range.unsatisfiable())
+ return HttpResponseModel.status(416);
+
+ if (range.partial()) {
+ int from = (int) range.start();
+ int to = (int) range.endInclusive();
+ byte[] slice = java.util.Arrays.copyOfRange(body, from, to + 1);
+ return HttpResponseModel.bytes(206, slice, contentType, disposition,
+ Map.of("Accept-Ranges", "bytes", "Content-Range", range.contentRange(body.length)));
+ }
+ return HttpResponseModel.bytes(200, body, contentType, disposition, Map.of("Accept-Ranges", "bytes"));
+ }
+
+ /**
+ * 构造文件响应。
+ *
+ * 完整下载必须回 200；命中 Range 才回 206 并带 {@code Content-Range}；
+ * 起点越界回 416。GET 与 HEAD 用同一套判定，HEAD 由外壳负责省略正文。
+ */
+ public static HttpResponseModel forFile(String rangeHeader, File file, String contentDispositionFileName,
+ String contentType) {
+ long length = file.length();
+ String disposition = contentDispositionFileName == null ? null : contentDisposition(contentDispositionFileName);
+ Map 目录名形如 {@code <标题> [ 磁盘访问经 {@link Supplier} 注入，测试可用内存清单替代真实文件系统。
+ */
+public final class GalleryArchiveIndex {
+
+ /** 返回画廊根目录下的一层子目录；可能为 null 或空。 */
+ private final Supplier 原 {@code findGalleryZipByGid} 对请求 gid 用 {@code lastIndexOf("[" + gid)} 定位，
+ * 并要求标记后的下一个字符是 {@code ']'} 或 {@code '-'}。由于查找的是完整的
+ * {@code "[ 与旧实现保持一致的判定：
+ *
+ * 这里不负责 Range 头解析与状态码 206/416/200 的选择：这些交给
+ * {@link FileResponseFactory}，以便同时被下载与订阅两条链路复用。
+ */
+@Slf4j
+public final class GalleryDownloadHandler implements HttpRequestHandler {
+
+ /** 旧下载链接没有 gid 时按路径中的文件名回到这个根目录下查找。 */
+ private static final String LEGACY_GALLERY_ROOT = "/root/gallery/gallery";
+
+ /** 兜底用的不存在路径：让「按文件名找不到归档」走统一的 404 分支。 */
+ private static final File FILE_NOT_FOUND = new File("/root/abc");
+
+ private final GalleryArchiveIndex archiveIndex;
+
+ public GalleryDownloadHandler(GalleryArchiveIndex archiveIndex) {
+ this.archiveIndex = archiveIndex;
+ }
+
+ @Override
+ public HttpResponseModel handle(HttpRequestModel request) {
+ if (!HttpRequestModel.GET.equals(request.method()) && !request.isHead())
+ return HttpResponseModel.text(501, "text/html", " 口令来自配置（环境变量 STORAGE_DOWNLOAD_ADMIN_CODE 优先，其次配置文件
+ * AdminDownloadCode）。未配置即关闭该能力；比较用固定时间算法。
+ */
+ public static boolean isAdminRequest(String authCode) {
+ String expected = Config.adminDownloadCode;
+ if (expected == null || expected.isBlank() || authCode == null)
+ return false;
+ return MessageDigest.isEqual(expected.getBytes(StandardCharsets.UTF_8),
+ authCode.getBytes(StandardCharsets.UTF_8));
+ }
+
+ /**
+ * 把管理员的请求路径解析为受配置根目录约束的文件。
+ *
+ * 用真实路径做前缀判断，因此 {@code ..} 与指向根目录之外的符号链接都会被拒。
+ * 文件不存在（含越界）返回 null，由调用方回 404。
+ */
+ public static File resolveAdminFile(String requestPath) {
+ if (requestPath == null || requestPath.isBlank())
+ return null;
+ try {
+ Path root = Path.of(Config.adminDownloadRoot).toRealPath();
+ // 兼容两种调用：绝对路径（主站历史上就是这么传的）与相对根目录的路径。
+ Path raw = Path.of(requestPath);
+ Path candidate = (raw.isAbsolute() ? raw : root.resolve(raw)).normalize();
+ if (!candidate.startsWith(root))
+ return null;
+ // 必须落到真实路径再判断一次：这样指向根目录之外的符号链接也会被拒。
+ Path resolved = candidate.toRealPath();
+ return resolved.startsWith(root) && resolved.toFile().isFile() ? resolved.toFile() : null;
+ } catch (Exception e) {
+ return null;
+ }
+ }
+}
diff --git a/src/main/java/lion/Http/HttpFileServer.java b/src/main/java/lion/Http/HttpFileServer.java
new file mode 100644
index 0000000..6e23535
--- /dev/null
+++ b/src/main/java/lion/Http/HttpFileServer.java
@@ -0,0 +1,261 @@
+package lion.Http;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * 基于 JDK {@code com.sun.net.httpserver.HttpServer} 的 HTTP 外壳。
+ *
+ * 取代原来的手工 {@code ServerSocket} 实现：请求行/请求头解析、HTTP/1.1 分帧与
+ * 连接保活交给 JDK，本类只负责把 {@link HttpExchange} 适配成
+ * {@link HttpRequestModel}，调用 {@link HttpRequestHandler}，再把
+ * {@link HttpResponseModel} 写回。
+ *
+ * 下载（8888）与订阅（8889）共用这一个外壳，只是注入的处理器不同。
+ */
+@Slf4j
+public final class HttpFileServer implements AutoCloseable {
+
+ /** 默认队列深度：请求超出线程数后先排队，队列满则退化为背压。 */
+ private static final int DEFAULT_QUEUE_DEPTH = 64;
+
+ private final HttpServer server;
+ private final ExecutorService executor;
+ private final int port;
+ private volatile boolean stopped;
+
+ private HttpFileServer(HttpServer server, ExecutorService executor, int port) {
+ this.server = server;
+ this.executor = executor;
+ this.port = port;
+ }
+
+ /**
+ * 绑定端口并创建上下文（尚未开始接收请求）。
+ *
+ * @param port 监听端口；0 表示由系统分配，便于测试
+ * @param path 上下文路径，例如 {@code "/"}
+ * @param handler 请求处理器
+ * @param workerCount 工作线程数上限
+ */
+ public static HttpFileServer create(int port, String path, HttpRequestHandler handler, int workerCount)
+ throws IOException {
+ return create(port, path, handler, workerCount, DEFAULT_QUEUE_DEPTH);
+ }
+
+ /**
+ * 绑定端口并创建上下文，显式指定排队深度。
+ *
+ * 执行器为有界 {@link ThreadPoolExecutor}：固定线程数 + {@link ArrayBlockingQueue}，
+ * 队列满时用 {@link ThreadPoolExecutor.CallerRunsPolicy} 在提交线程（JDK HttpServer 的
+ * dispatcher）上直接执行，形成对 accept 的背压，而不是无上限堆积任务或静默丢弃请求。
+ */
+ public static HttpFileServer create(int port, String path, HttpRequestHandler handler,
+ int workerCount, int queueDepth) throws IOException {
+ Objects.requireNonNull(handler, "handler");
+ int workers = Math.max(1, workerCount);
+ // backlog 只影响内核半连接队列，取一个足够大的固定值即可。
+ HttpServer server = HttpServer.create(new InetSocketAddress(port), 128);
+ ExecutorService executor = new ThreadPoolExecutor(
+ workers, workers, 0L, TimeUnit.MILLISECONDS,
+ new ArrayBlockingQueue<>(Math.max(1, queueDepth)),
+ namedDaemonThreads("storage-http"),
+ new ThreadPoolExecutor.CallerRunsPolicy());
+ server.setExecutor(executor);
+ server.createContext(path, exchange -> serve(exchange, handler));
+ return new HttpFileServer(server, executor, server.getAddress().getPort());
+ }
+
+ /** 开始接收请求。 */
+ public void start() {
+ server.start();
+ }
+
+ /** 实际绑定到的端口（{@code create} 传 0 时用于取回随机端口）。 */
+ public int port() {
+ return port;
+ }
+
+ /** 执行器，供包内测试校验有界队列与背压策略。 */
+ ExecutorService executor() {
+ return executor;
+ }
+
+ /**
+ * 停止接收请求并释放线程。
+ *
+ * {@code HttpServer.stop} 最多等待 1 秒让在途交换结束，随后强制关闭连接；
+ * 线程池在本方法内收敛，可安全重复调用。
+ */
+ @Override
+ public void close() {
+ if (stopped)
+ return;
+ stopped = true;
+ try {
+ server.stop(1);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ /** 单次请求处理：适配 → 路由 → 写响应；任何异常都不得打死外壳。 */
+ private static void serve(HttpExchange exchange, HttpRequestHandler handler) {
+ try (exchange) {
+ HttpResponseModel response;
+ try {
+ response = handler.handle(toRequestModel(exchange));
+ } catch (Exception failure) {
+ log.error("处理请求失败: {} {}", exchange.getRequestMethod(), exchange.getRequestURI(), failure);
+ response = HttpResponseModel.text(500, "text/html", " 路径不是 {@code application/x-www-form-urlencoded}，{@code +} 在路径里就是普通字符
+ * （例如文件名 {@code plus+name.zip}），不能按空格处理。这里先把 {@code +} 预转义成
+ * {@code %2B} 再交给 {@link URLDecoder}，从而只保留其百分号解码能力。
+ */
+ public static String decodePath(String value) {
+ if (value == null)
+ return null;
+ return decodeQuery(value.replace("+", "%2B"));
+ }
+
+ /** 查询串解码：标准表单语义，{@code +} 表示空格。非法序列退回原文。 */
+ public static String decodeQuery(String value) {
+ if (value == null)
+ return null;
+ try {
+ return URLDecoder.decode(value, StandardCharsets.UTF_8);
+ } catch (IllegalArgumentException malformed) {
+ return value;
+ }
+ }
+
+ /**
+ * 写响应。
+ *
+ * HEAD 必须显式写 {@code Content-Length} 并以 {@code -1} 结束，否则 JDK 实现
+ * 不会给出长度；任何响应都不写正文，避免客户端读到错位的数据。
+ */
+ static void write(HttpExchange exchange, HttpResponseModel response) throws IOException {
+ boolean head = "HEAD".equalsIgnoreCase(exchange.getRequestMethod());
+ long bodyLength = response.hasFile() ? response.bodyLength()
+ : (response.body() == null ? 0 : response.body().length);
+
+ var headers = exchange.getResponseHeaders();
+ if (response.contentType() != null)
+ headers.set("Content-Type", response.contentType());
+ if (response.contentDisposition() != null)
+ headers.set("Content-Disposition", response.contentDisposition());
+ for (Map.Entry 在 JDK {@code HttpServer} 的 {@code HttpExchange} 之上再抽一层，使路由与业务判定
+ * 可以被单元测试直接驱动，不必每次都起真实端口；真实集成测试仍走 HTTP。
+ */
+public final class HttpRequestModel {
+
+ /** {@code GET}，唯一允许下载的方法。 */
+ public static final String GET = "GET";
+ /** {@code HEAD}，与 GET 同语义但不回正文。 */
+ public static final String HEAD = "HEAD";
+
+ private final String method;
+ private final String rawPath;
+ private final String decodedPath;
+ private final Map 用于来源 IP 不在白名单的请求：与旧实现一致，既不泄露服务存在，也不给出可探测
+ * 的差异。这不属于 HTTP 响应，外壳必须直接断开。
+ */
+ public static HttpResponseModel rejected() {
+ return new HttpResponseModel(0, null, null, 0, 0, null, null, Map.of(), true);
+ }
+
+ /** 文件实体；{@code start}/{@code length} 描述要发送的字节区间。 */
+ public static HttpResponseModel file(int status, File file, long start, long length,
+ String contentType, String contentDisposition,
+ Map {@code messageType}/{@code messageId} 一直是公开字段，这里补上标准的 getter/setter，
+ * 让两端测试都能用属性访问。这些访问器生成的 JSON 属性名与原有字段名完全相同，
+ * 因此线上字节没有任何变化。
+ *
+ * 刻意不用 Lombok：{@code @Data} 会为继承它的消息类生成 equals/hashCode/toString，
+ * 改变既有消息的相等语义。这里只补最小必要的访问器。
+ */
public class AbstractMessage {
public static final byte DOWNLOAD_POST_MESSAGE = 1;
@@ -18,7 +28,35 @@ public class AbstractMessage {
public static final byte SUBSCRIPTION_SNAPSHOT_MESSAGE = 9;
+ /** 协议能力协商（v2，见 protocol-v2.md）。 */
+ public static final byte PROTOCOL_HELLO_MESSAGE = 10;
+
+ /** 可靠任务命令（v2）：稳定 operationId 的 DOWNLOAD/DELETE。 */
+ public static final byte TASK_COMMAND_MESSAGE = 11;
+
+ /** 可靠任务报告（v2）：带 reportId/revision 的不可变快照。 */
+ public static final byte TASK_REPORT_MESSAGE = 12;
+
+ /** 可靠任务报告确认（v2）。 */
+ public static final byte TASK_REPORT_ACK_MESSAGE = 13;
+
public byte messageType;
public int messageId;
+
+ public byte getMessageType() {
+ return messageType;
+ }
+
+ public void setMessageType(byte messageType) {
+ this.messageType = messageType;
+ }
+
+ public int getMessageId() {
+ return messageId;
+ }
+
+ public void setMessageId(int messageId) {
+ this.messageId = messageId;
+ }
}
diff --git a/src/main/java/lion/Message/Main/ProtocolHelloMessage.java b/src/main/java/lion/Message/Main/ProtocolHelloMessage.java
new file mode 100644
index 0000000..054a13b
--- /dev/null
+++ b/src/main/java/lion/Message/Main/ProtocolHelloMessage.java
@@ -0,0 +1,49 @@
+package lion.Message.Main;
+
+import lion.Message.AbstractMessage;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 协议能力协商（type=10）。
+ *
+ * 主站在身份消息之后发送本消息；节点收到 {@code protocolVersion >= 2} 后回一条同样的
+ * hello，随后才允许使用 type 11/12/13 的可靠任务消息。
+ *
+ * 字段刻意保持最小：旧节点/旧主站的 Jackson 会拒绝未知字段，因此旧消息类型一律不得
+ * 增加字段，新增能力只能通过新类型承载。
+ */
+@Data
+@NoArgsConstructor
+public class ProtocolHelloMessage extends AbstractMessage {
+ {
+ messageType = PROTOCOL_HELLO_MESSAGE;
+ }
+
+ /** 当前节点实现的协议版本。 */
+ public static final int VERSION = 2;
+
+ /** 可靠任务协议的默认能力标记。 */
+ public static final String RELIABLE_TASK_V2 = "reliable-task-v2";
+
+ private int protocolVersion;
+
+ private List {@code operationId} 对同一逻辑操作稳定不变：主站重试、断线重发必须复用同一个 id。
+ * 节点按 operationId 去重，绝不重复执行已成功的副作用。
+ */
+@Data
+@NoArgsConstructor
+public class TaskCommandMessage extends AbstractMessage {
+ {
+ messageType = TASK_COMMAND_MESSAGE;
+ }
+
+ public static final String ACTION_DOWNLOAD = "DOWNLOAD";
+ public static final String ACTION_DELETE = "DELETE";
+
+ private String operationId;
+
+ /** {@link #ACTION_DOWNLOAD} 或 {@link #ACTION_DELETE}。 */
+ private String action;
+
+ private int gid;
+
+ private String name;
+
+ /**
+ * 必须 {@link JsonIgnore}：Jackson 会把 {@code isDownload()} 当成属性 {@code download} 一并
+ * 序列化，而对端主站使用更严格的 Jackson，默认拒绝未知字段，握手会直接失败。
+ */
+ @JsonIgnore
+ public boolean isDownload() {
+ return ACTION_DOWNLOAD.equals(action);
+ }
+
+ @JsonIgnore
+ public boolean isDelete() {
+ return ACTION_DELETE.equals(action);
+ }
+}
diff --git a/src/main/java/lion/Message/Main/TaskReportAckMessage.java b/src/main/java/lion/Message/Main/TaskReportAckMessage.java
new file mode 100644
index 0000000..290dbce
--- /dev/null
+++ b/src/main/java/lion/Message/Main/TaskReportAckMessage.java
@@ -0,0 +1,36 @@
+package lion.Message.Main;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import lion.Message.AbstractMessage;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * 可靠任务报告的确认（type=13，主站 → 节点）。
+ *
+ * 主站只在报告落库并提交后回 {@link #RESULT_SUCCESS}；落库失败不得回成功。节点收到成功
+ * ACK 才允许移除对应版本的快照，收到非成功结果则保留并重试。
+ */
+@Data
+@NoArgsConstructor
+public class TaskReportAckMessage extends AbstractMessage {
+ {
+ messageType = TASK_REPORT_ACK_MESSAGE;
+ }
+
+ /** 报告已提交落库，节点可安全移除该版本快照。 */
+ public static final byte RESULT_SUCCESS = 0;
+
+ /** 主站未提交（例如数据库失败），节点必须保留快照并重试。 */
+ public static final byte RESULT_REJECTED = 1;
+
+ private String reportId;
+
+ private byte result;
+
+ /** 同 {@link TaskCommandMessage#isDownload()}：避免被序列化成未知字段 {@code success}。 */
+ @JsonIgnore
+ public boolean isSuccess() {
+ return result == RESULT_SUCCESS;
+ }
+}
diff --git a/src/main/java/lion/Message/Main/TaskReportMessage.java b/src/main/java/lion/Message/Main/TaskReportMessage.java
new file mode 100644
index 0000000..4535116
--- /dev/null
+++ b/src/main/java/lion/Message/Main/TaskReportMessage.java
@@ -0,0 +1,33 @@
+package lion.Message.Main;
+
+import lion.Domain.GalleryTask;
+import lion.Message.AbstractMessage;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * 可靠任务状态报告（type=12，节点 → 主站）。
+ *
+ * 每条报告有稳定的 {@code reportId} 与单调递增的 {@code revision}。主站按 reportId 幂等
+ * 落库、在提交后回 {@link TaskReportAckMessage}；节点只有在收到匹配版本的 ACK 后才移除对应
+ * 快照。重发必须复用同一 reportId 与 revision。
+ */
+@Data
+@NoArgsConstructor
+public class TaskReportMessage extends AbstractMessage {
+ {
+ messageType = TASK_REPORT_MESSAGE;
+ }
+
+ private String reportId;
+
+ private long revision;
+
+ private GalleryTask[] galleryTasks;
+
+ /** gid → 该任务的活跃操作 operationId，供主站丢弃过期或已删除任务的报告。 */
+ private Map HTTP 外壳已改为 JDK {@code com.sun.net.httpserver.HttpServer}（见
+ * {@link HttpFileServer}），本类只保留启动逻辑、来源 IP 判定，以及供既有测试使用的
+ * 兼容入口。端口仍为 {@value #PORT}。
+ */
@Slf4j
-public class MultiThreadedHTTPServer {
- private static final int PORT = 8888;
+public final class MultiThreadedHTTPServer {
- /** 已安装的客户端可能连上却不发请求；没有读超时就会一直占住工作线程。 */
- private static final int SOCKET_READ_TIMEOUT_MILLIS = 30_000;
+ /** 对外下载端口；改动会同时影响 Nginx 转发配置，保持固定。 */
+ public static final int PORT = 8888;
- /** 兜底用的不存在路径：让「按文件名找不到归档」走统一的 404 分支。 */
- private static final File FILE_NOT_FOUND = new File("/root/abc");
+ /** 工作线程上限，避免无界线程池在异常流量下耗尽资源。 */
+ private static final int WORKER_THREADS = 16;
+ /** 未命中后允许重扫归档目录的最小间隔，让新归档可见但不会被未知 gid 拖成扫描风暴。 */
+ private static final long ARCHIVE_INDEX_RESCAN_MILLIS = 5_000L;
+
+ private static final String GALLERY_ROOT = "/root/gallery/gallery";
+
+ private MultiThreadedHTTPServer() {
+ }
+
+ /**
+ * 兼容旧入口：在独立线程里启动服务并阻塞，直到当前线程被中断。
+ *
+ * 历史上由 {@code Main} 用 {@code new Thread(() -> main(null))} 调用，因此这里
+ * 不能立即返回；同时不能阻止 JVM 退出，故服务线程全部为守护线程。
+ */
public static void main(String[] args) {
- ExecutorService threadPool = Executors.newCachedThreadPool();
- String real_ip;
- try{
- real_ip = InetAddress.getByName("lionwebsite.xyz").getHostAddress();
- } catch (UnknownHostException ignored){
- real_ip = "207.60.50.74";
+ String realIp;
+ try {
+ realIp = InetAddress.getByName("lionwebsite.xyz").getHostAddress();
+ } catch (UnknownHostException ignored) {
+ realIp = "207.60.50.74";
}
- // accept 循环必须扛得住单次失败：否则一次异常就让下载服务整体退出，
- // 之后所有下载都 404，直到有人手工重启。
- while (!Thread.currentThread().isInterrupted()) {
- try (ServerSocket serverSocket = new ServerSocket(PORT)) {
- log.info("Server listening on port {}", PORT);
- while (true) {
- Socket clientSocket = serverSocket.accept();
- 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());
+
+ try (HttpFileServer server = createServer(realIp)) {
+ server.start();
+ log.info("下载服务已启动，监听端口 {}", server.port());
+ while (!Thread.currentThread().isInterrupted()) {
try {
- Thread.sleep(5_000);
+ Thread.sleep(60_000);
} catch (InterruptedException interrupted) {
Thread.currentThread().interrupt();
}
}
- }
- }
-
- // 包级可见：与 parseRequestLine 一样留作测试入口，便于对真实 socket 断言请求处理。
- static void handleClientRequest(Socket clientSocket) {
- String fileName = "";
- try (Socket socket = clientSocket) {
- // 没有读超时的连接可以永远占住一个工作线程，必须设一个上限。
- socket.setSoTimeout(SOCKET_READ_TIMEOUT_MILLIS);
- BufferedReader requestReader = new BufferedReader(
- new InputStreamReader(socket.getInputStream(), StandardCharsets.ISO_8859_1));
- String requestLine = requestReader.readLine();
-
- // 空行或没有空格的请求行会让 requestParts[1] 越界。这种异常不是 IOException，
- // 原样抛出会被线程池静默吞掉、socket 也关不掉，因此在这里显式判非法。
- if (requestLine == null || requestLine.isBlank())
- return;
-
- // Parse the request line to get the method and path
- String[] requestParts = requestLine.trim().split("\\s+");
- if (requestParts.length < 2) {
- CustomUtil.sendErrorResponse(socket, "400 Bad Request");
- return;
- }
- String method = requestParts[0];
- Map 口令不再写死在源码里，改为配置（环境变量 STORAGE_DOWNLOAD_ADMIN_CODE 优先，
- * 其次配置文件 AdminDownloadCode）。未配置即关闭该能力；比较用固定时间算法，
- * 避免按前缀长度泄露口令。
+ * @param allowedClientIp 唯一允许的来源 IP；为 null 时不限制来源
*/
- static boolean isAdminRequest(String authCode){
- String expected = Config.adminDownloadCode;
- if (expected == null || expected.isBlank() || authCode == null)
- return false;
- return MessageDigest.isEqual(expected.getBytes(StandardCharsets.UTF_8),
- authCode.getBytes(StandardCharsets.UTF_8));
+ public static HttpFileServer createServer(String allowedClientIp) throws IOException {
+ GalleryArchiveIndex archiveIndex = new GalleryArchiveIndex(
+ new File(GALLERY_ROOT), ARCHIVE_INDEX_RESCAN_MILLIS);
+ HttpRequestHandler handler = withClientFilter(new GalleryDownloadHandler(archiveIndex), allowedClientIp);
+ return HttpFileServer.create(PORT, "/", handler, WORKER_THREADS);
}
/**
- * 把管理员的请求路径解析为受配置根目录约束的文件。
+ * 在业务处理器之前插入来源校验：非白名单 IP 直接关闭连接。
*
- * 用真实路径做前缀判断，因此 {@code ..} 与指向根目录之外的符号链接都会被拒。
- * 文件不存在（含越界）返回 null，由调用方回 404。
+ * 与旧实现一致：被拒的连接不返回任何响应，避免向未知来源暴露服务存在。
+ * 对端 IP 取不到时（例如本地测试直接驱动、或代理未传）默认放行，行为与旧实现
+ * 只在「拿到 IP 且不等于白名单」时拒绝相同。
*/
- static File resolveAdminFile(String requestPath){
- if (requestPath == null || requestPath.isBlank())
- return null;
- try {
- Path root = Path.of(Config.adminDownloadRoot).toRealPath();
- // 兼容两种调用：绝对路径（主站历史上就是这么传的）与相对根目录的路径。
- Path raw = Path.of(requestPath);
- Path candidate = (raw.isAbsolute() ? raw : root.resolve(raw)).normalize();
- if (!candidate.startsWith(root))
- return null;
- // 必须落到真实路径再判断一次：这样指向根目录之外的符号链接也会被拒。
- Path resolved = candidate.toRealPath();
- return resolved.startsWith(root) && resolved.toFile().isFile() ? resolved.toFile() : null;
- } catch (Exception e) {
- return null;
- }
+ static HttpRequestHandler withClientFilter(HttpRequestHandler delegate, String allowedClientIp) {
+ return request -> {
+ String clientIp = request.clientAddress();
+ if (allowedClientIp != null && clientIp != null && !allowedClientIp.equals(clientIp)) {
+ log.info("拒绝未知来源: {}", clientIp);
+ return HttpResponseModel.rejected();
+ }
+ return delegate.handle(request);
+ };
}
- private static File findGalleryZipByGid(File galleryRoot, String gid){
- File[] galleryDirectories = galleryRoot.listFiles(File::isDirectory);
- if(galleryDirectories == null)
- return null;
+ // ---- 以下为既有测试依赖的兼容入口，转发到 lion.Http 的实现 ----
- String marker = "[" + gid;
- 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;
+ /** @see GalleryDownloadHandler#isAdminRequest(String) */
+ static boolean isAdminRequest(String authCode) {
+ return GalleryDownloadHandler.isAdminRequest(authCode);
}
+ /** @see GalleryDownloadHandler#resolveAdminFile(String) */
+ static File resolveAdminFile(String requestPath) {
+ return GalleryDownloadHandler.resolveAdminFile(requestPath);
+ }
+
+ /**
+ * 解析请求行中的路径与查询参数。
+ *
+ * @return 含 {@code path} 的映射；请求行为 null 时返回 null
+ */
public static Map 删除命令与压缩线程会争用同一份画廊目录：删除请求若在压缩写入归档的过程中
+ * 移走源目录，会留下半个归档或让压缩失败。这里的约定是删除先查该集合，命中即回
+ * 可重试的「忙」，等压缩结束后由主站重试，而不是并发破坏数据。
+ */
+ final Set 删除与压缩是互斥的：命中时调用方应回可重试的「忙」，而不是继续动磁盘。
+ */
+ public boolean isCompressing(int gid){
+ if (activeCompressions.contains(gid))
+ return true;
+ synchronized (compress_queue) {
+ for (GalleryTask queued : compress_queue)
+ if (queued.getGid() == gid)
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * 从压缩队列与任务队列中摘除该 gid。
+ *
+ * 用于删除成功后清理本节点内部状态，避免已删除任务继续被扫描或压缩。
+ */
+ public void removeGid(int gid){
+ queue.remove(gid);
+ retryAfter.remove(gid);
+ synchronized (compress_queue) {
+ compress_queue.removeIf(task -> task.getGid() == gid);
+ }
+ }
+
+ /**
+ * 关闭本服务持有的线程资源。
+ *
+ * 压缩调度器与通知线程都是守护线程，进程退出不会因此挂住；但测试与将来可能的
+ * 热重启必须能显式释放，否则每个实例都会留下一个仍在跑压缩的线程。
+ */
+ public void shutdown(){
+ if (convert_thread != null)
+ convert_thread.shutdownNow();
+ notificationExecutor.shutdownNow();
+ }
+
static boolean isValidArchive(File file) {
if (!file.isFile())
return false;
@@ -281,6 +337,20 @@ public class DownloadCheckService {
return null;
}
+ /**
+ * 按 gid 在存储目录中解析真实归档目录名。
+ *
+ * 删除命令需要确切目录名：主站下发的 name 可能带下载器追加的分辨率后缀，直接用它会
+ * 删错对象或报找不到。先按 gid 找已归档目录，再回退到下载目录，最后才用主站给的名字。
+ */
+ public String findDirectoryNameByGid(int gid){
+ File stored = findStoredDirectoryByGid(gid);
+ if (stored != null)
+ return stored.getName();
+ File downloading = findDirectoryByGid(new File(downloadPath), gid);
+ return downloading == null ? null : downloading.getName();
+ }
+
/**
* 压缩包校验带缓存：归档按大小不变，校验结果可复用。
* 大小变化（重新压缩）或距上次校验超过一小时才重算，避免重复整包读取。
diff --git a/src/main/java/lion/Service/NodeErrorCode.java b/src/main/java/lion/Service/NodeErrorCode.java
new file mode 100644
index 0000000..d3a4c5d
--- /dev/null
+++ b/src/main/java/lion/Service/NodeErrorCode.java
@@ -0,0 +1,34 @@
+package lion.Service;
+
+/**
+ * 节点侧命令结果码。
+ *
+ * 前两个值与主站 {@code Error.ErrorCode} 保持一致，直接复用旧协议的语义；
+ * {@link #BUSY_COMPRESSING} 是可靠任务协议新增的可重试拒绝：目标正在压缩，此刻删除或
+ * 覆盖会把半成品归档暴露出去，主站应以同一 operationId 稍后重试，节点不会记为永久成功。
+ */
+public class NodeErrorCode {
+
+ /** 命令成功。 */
+ public static final byte SUCCESS = 0;
+
+ /** 立即重试不一定成功，但状态可能变化，属可重试失败。 */
+ public static final byte IO_ERROR = 1;
+
+ /** 目标不存在。对删除而言等价于「已达成目标」，是终态。 */
+ public static final byte FILE_NOT_FOUND = 2;
+
+ /** 目标正在压缩，当前无法处理；可重试，不得记为永久结果。 */
+ public static final byte BUSY_COMPRESSING = 3;
+
+ /** 命令本身不合法（缺 operationId 或未知 action），重试不会改变结果。 */
+ public static final byte INVALID_COMMAND = 4;
+
+ /** 终态结果才写入回执，重试直接回放，不再执行副作用。 */
+ public static boolean isTerminal(byte result) {
+ return result == SUCCESS || result == FILE_NOT_FOUND || result == INVALID_COMMAND;
+ }
+
+ private NodeErrorCode() {
+ }
+}
diff --git a/src/main/java/lion/Service/ReliableTaskService.java b/src/main/java/lion/Service/ReliableTaskService.java
new file mode 100644
index 0000000..2df2241
--- /dev/null
+++ b/src/main/java/lion/Service/ReliableTaskService.java
@@ -0,0 +1,296 @@
+package lion.Service;
+
+import lion.Domain.GalleryTask;
+import lion.Message.Main.TaskCommandMessage;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * 可靠任务协议在节点侧的执行与状态编排。
+ *
+ * 它把三样东西绑在一起：任务队列（运行期事实）、{@link TaskJournal}（持久事实）、
+ * {@link TaskReporter}（上报与 ACK 生命周期）。命令先查回执再执行、成功副作用先落盘再回执；
+ * 报告先落盘再发送、只有匹配版本的 ACK 才清理任务。
+ *
+ * 所有对日志与队列的修改都在本类的锁下完成，保证「落盘」与「内存状态」不会分叉。
+ */
+@Slf4j
+public class ReliableTaskService {
+
+ private final TaskJournal journal;
+
+ private final TaskReporter reporter;
+
+ private final DownloadCheckService downloadCheckService;
+
+ /** 任务队列与日志共享的真实来源；节点扫描会就地更新其中的对象。 */
+ private final Map 先读回日志，再把它合并进内存队列；被中断的压缩一律退回「下载完成」，让扫描与压缩
+ * 重新评估真实文件，而不是停在一个永远等不到结果的中间态。真实文件状态由后续扫描重算，
+ * 日志里的进度只作为扫描前的乐观初值。
+ */
+ public void restore() throws IOException {
+ journal.load();
+ synchronized (lock) {
+ for (TaskJournal.JournalTask stored : journal.tasks().values()) {
+ GalleryTask task = stored.toGalleryTask();
+ if (task.getStatus() == GalleryTask.COMPRESSING)
+ task.setStatus(GalleryTask.DOWNLOAD_COMPLETE);
+ queue.putIfAbsent(task.getGid(), task);
+ if (stored.operationId != null)
+ operationIds.put(stored.gid, stored.operationId);
+ }
+ log.info("节点重启恢复: 任务={}, 待 ACK 报告={}", queue.size(), journal.pendingReports().size());
+ }
+ }
+
+ /**
+ * 执行一条可靠任务命令。
+ *
+ * 返回的字节就是回给主站的结果码。已终态成功的 operationId 直接回放原结果，
+ * 不再触碰磁盘或队列；可重试失败不写回执，主站以同一 operationId 重试即可。
+ */
+ public byte handleCommand(TaskCommandMessage command) {
+ if (command == null || command.getOperationId() == null || command.getOperationId().isBlank()
+ || command.getAction() == null)
+ return NodeErrorCode.INVALID_COMMAND;
+
+ synchronized (lock) {
+ TaskJournal.CommandReceipt receipt = journal.receipt(command.getOperationId());
+ if (receipt != null) {
+ log.info("命令已执行过，回放回执: operationId={}, result={}", command.getOperationId(), receipt.result);
+ return receipt.result;
+ }
+
+ try {
+ if (command.isDownload())
+ return executeDownload(command);
+ if (command.isDelete())
+ return executeDelete(command);
+ return NodeErrorCode.INVALID_COMMAND;
+ } catch (IOException e) {
+ // 落盘失败时绝不能把副作用当成成功：回可重试错误，主站会以同一 id 再来。
+ log.error("可靠命令落盘失败，回可重试错误: operationId={}", command.getOperationId(), e);
+ return NodeErrorCode.IO_ERROR;
+ }
+ }
+ }
+
+ private byte executeDownload(TaskCommandMessage command) throws IOException {
+ GalleryTask incoming = new GalleryTask();
+ incoming.setGid(command.getGid());
+ incoming.setName(command.getName());
+
+ GalleryTask active = downloadCheckService.addToQueue(incoming);
+ operationIds.put(active.getGid(), command.getOperationId());
+ queue.putIfAbsent(active.getGid(), active);
+
+ // 先把入队事实写进日志，再写回执：回执一旦落盘就代表「已受理」，重启后必须还能看到任务。
+ journal.bumpAndPutTask(active, command.getOperationId());
+ journal.putReceipt(TaskJournal.CommandReceipt.of(
+ command.getOperationId(), TaskCommandMessage.ACTION_DOWNLOAD,
+ active.getGid(), active.getName(), NodeErrorCode.SUCCESS));
+ return NodeErrorCode.SUCCESS;
+ }
+
+ private byte executeDelete(TaskCommandMessage command) throws IOException {
+ int gid = command.getGid();
+
+ // 压缩中删除会破坏正在进行中的归档写入，先回可重试的「忙」，不写回执。
+ if (downloadCheckService.isCompressing(gid)) {
+ log.info("任务正在压缩，删除请求延后重试: gid={}", gid);
+ return NodeErrorCode.BUSY_COMPRESSING;
+ }
+
+ String directoryName = resolveDirectoryName(gid, command.getName());
+ byte result = directoryName == null
+ ? NodeErrorCode.FILE_NOT_FOUND
+ : deleteGallery(directoryName);
+
+ if (!NodeErrorCode.isTerminal(result))
+ return result;
+
+ downloadCheckService.removeGid(gid);
+ queue.remove(gid);
+ operationIds.remove(gid);
+ journal.removeTask(gid);
+ journal.putReceipt(TaskJournal.CommandReceipt.of(
+ command.getOperationId(), TaskCommandMessage.ACTION_DELETE,
+ gid, directoryName, result));
+ return result;
+ }
+
+ private byte deleteGallery(String name) {
+ File target = storagePath.resolve(name).toFile();
+ if (!target.isDirectory())
+ return NodeErrorCode.FILE_NOT_FOUND;
+ return DeleteService.deleteWithin(storagePath.toString(), name);
+ }
+
+ /** 优先按 gid 解析真实目录名；主站的名称可能因下载器追加后缀而失真。 */
+ private String resolveDirectoryName(int gid, String fallbackName) {
+ String byGid = downloadCheckService.findDirectoryNameByGid(gid);
+ if (byGid != null)
+ return byGid;
+ if (fallbackName != null && !fallbackName.isBlank()
+ && storagePath.resolve(fallbackName).toFile().isDirectory())
+ return fallbackName;
+ return null;
+ }
+
+ /**
+ * 把扫描后的队列状态对齐到日志。返回是否产生了需要上报的变化。
+ */
+ public boolean refreshJournalFromQueue() throws IOException {
+ synchronized (lock) {
+ return journal.syncTasks(List.copyOf(queue.values()), operationIds);
+ }
+ }
+
+ /**
+ * 生成一份待发送报告并先落盘。无任务可报时返回空。
+ *
+ * 若当前日志中的任务版本已被某份待 ACK 快照完整覆盖，则不重复生成：断线重连时应当
+ * 原样重发旧快照，而不是为同一批内容再堆一份新快照，否则主站要处理无限增长的重复报告。
+ */
+ public Optional ACK 只代表「主站已落库收到这份报告」，不代表任务已经完成。因此这里只移除真正的终态
+ * 任务（{@link GalleryTask#COMPRESS_COMPLETE}）；处于下载中、压缩中等进行中的任务即使被
+ * 确认，也必须留在队列与日志里继续推进，否则会在下载尚未结束时被提前遗忘。
+ *
+ * 还要额外比对运行期状态与快照内容：扫描线程在锁外就地更新 {@link GalleryTask}，
+ * 一份迟到的 ACK 可能在任务又前进之后才到。若只比 revision，扫描尚未落盘的那次变化
+ * 会让旧 ACK 误删更新过的任务，所以状态与进度也必须一致才允许移除。
+ *
+ * 返回被移除的 gid；重复、失败、乱序或非终态的 ACK 不产生任何移除。
+ */
+ public Set 它承担协议 v2 的「可恢复」部分：任务、命令回执、单调递增的 revision，以及尚未收到
+ * ACK 的报告快照。所有写入都先落盘再对内存生效——节点在写盘前崩溃，主站看不到任何未落盘
+ * 的副作用；写盘后才崩溃，重启时能从磁盘恢复待重发的报告。
+ *
+ * 并发约定：本类的方法不是线程安全的，调用方（{@link ReliableTaskService}）必须在自己的
+ * 锁下调用。这样做是为了让「持久化 + 内存状态更新」在同一个临界区内保持一致。
+ */
+@Slf4j
+public class TaskJournal {
+
+ /** 日志文件固定名，便于运维定位与备份。 */
+ static final String STATE_FILE = "task-journal.json";
+
+ /** 落盘格式版本，供后续不兼容变更识别。 */
+ static final int SCHEMA_VERSION = 1;
+
+ private final Path directory;
+
+ private final ObjectMapper objectMapper;
+
+ /**
+ * 当前已落盘的状态。
+ *
+ * 内存状态只在成功写盘之后才被替换，因此任何一次 I/O 失败都不会让内存领先磁盘：
+ * 调用方拿到 {@link IOException} 时，它与磁盘看到的仍是同一份数据。若允许「先改内存再写盘」，
+ * 写盘失败会留下「内存已受理、磁盘没有」的裂缝，重启后命令回执与任务全部消失，
+ * 主站重试会重复执行副作用。
+ */
+ private volatile State state = new State();
+
+ public TaskJournal(Path directory) {
+ this(directory, CustomUtil.objectMapper);
+ }
+
+ TaskJournal(Path directory, ObjectMapper objectMapper) {
+ this.directory = directory;
+ this.objectMapper = objectMapper;
+ }
+
+ /**
+ * 从磁盘恢复。目录不存在时视为空日志，并交由调用方决定是否创建。
+ *
+ * 文件损坏（截断、非法 JSON）不会被静默吞掉：抛出异常让启动失败，避免带着一份
+ * 空日志继续运行、把主站上「已完成」的任务重新看成未完成而重复上报或重复执行。
+ */
+ public void load() throws IOException {
+ Path file = directory.resolve(STATE_FILE);
+ if (!Files.isRegularFile(file)) {
+ log.info("任务日志不存在，以空日志启动: {}", file);
+ return;
+ }
+ State loaded;
+ try {
+ loaded = objectMapper.readValue(Files.readAllBytes(file), State.class);
+ } catch (IOException e) {
+ throw new IOException("任务日志损坏，拒绝以不一致状态启动: " + file, e);
+ }
+ // 版本校验必须发生在使用数据之前：更高版本的日志可能含有本程序不认识的语义
+ // （例如新的任务字段或状态），按旧规则解释会静默丢字段。宁可拒绝启动。
+ if (loaded.schemaVersion > SCHEMA_VERSION) {
+ throw new IOException("任务日志 schemaVersion=" + loaded.schemaVersion
+ + " 高于本节点支持的 " + SCHEMA_VERSION + "，拒绝启动: " + file);
+ }
+ if (loaded.schemaVersion < SCHEMA_VERSION) {
+ log.warn("任务日志 schemaVersion={} 低于当前 {}，按兼容方式读取",
+ loaded.schemaVersion, SCHEMA_VERSION);
+ }
+ state.schemaVersion = loaded.schemaVersion;
+ state.revision = loaded.revision;
+ state.tasks = loaded.tasks == null ? new LinkedHashMap<>() : new LinkedHashMap<>(loaded.tasks);
+ state.commandReceipts = loaded.commandReceipts == null ? new LinkedHashMap<>() : new LinkedHashMap<>(loaded.commandReceipts);
+ state.pendingReports = loaded.pendingReports == null ? new LinkedHashMap<>() : new LinkedHashMap<>(loaded.pendingReports);
+ log.info("任务日志已恢复: tasks={}, receipts={}, pendingReports={}, revision={}",
+ state.tasks.size(), state.commandReceipts.size(), state.pendingReports.size(), state.revision);
+ }
+
+ // ---- 任务 ----
+
+ public Map revision 与任务记录必须在同一次落盘里生效：若分两次写，节点可能在「revision 已涨
+ * 但任务还没更新」的状态下崩溃，重启后该任务的变化会被判定为已上报而永久丢失。
+ */
+ public JournalTask bumpAndPutTask(GalleryTask task, String operationId) throws IOException {
+ final JournalTask[] created = new JournalTask[1];
+ commit(next -> {
+ JournalTask journalTask = JournalTask.of(task, operationId, next.revision + 1);
+ next.revision = journalTask.revision;
+ next.tasks.put(journalTask.gid, journalTask);
+ created[0] = journalTask;
+ });
+ return created[0];
+ }
+
+ /** 仅当当前记录版本与期望一致时才移除，避免删掉更新的任务。 */
+ public boolean removeTaskIfRevision(int gid, long revision) throws IOException {
+ JournalTask current = state.tasks.get(gid);
+ if (current == null || current.revision != revision)
+ return false;
+ commit(next -> next.tasks.remove(gid));
+ return true;
+ }
+
+ public void removeTask(int gid) throws IOException {
+ if (state.tasks.containsKey(gid))
+ commit(next -> next.tasks.remove(gid));
+ }
+
+ /**
+ * 用运行期任务队列对齐日志，一次落盘完成全部变化。
+ *
+ * 把「扫描出的新状态」批量写进日志，而不是每个任务各写一次：每轮扫描可能同时更新
+ * 多个任务的进度，逐条 flush 会让磁盘写入量随任务数线性放大。所有变化共享同一个新
+ * revision，这也让一份报告能一次覆盖它们。
+ *
+ * @return 是否发生了需要上报的变化
+ */
+ public boolean syncTasks(Collection 这是「I/O 失败不得让内存领先磁盘」的实现方式。{@code mutation} 抛异常时状态原样保留；
+ * 写盘抛异常时副本被丢弃，内存仍是最后一次成功落盘的内容。调用方在捕获 {@link IOException}
+ * 后可以安全地把操作当作「未发生」重试，而不会重复执行已生效的副作用。
+ */
+ private void commit(Consumer 必须逐层复制到可变条目：只拷 map 的外层会让「新状态」与「旧状态」共享同一个
+ * JournalTask 实例，写盘失败后旧状态里的对象也已被改写，回滚就失效了。
+ */
+ State copy() {
+ State clone = new State();
+ clone.schemaVersion = schemaVersion;
+ clone.revision = revision;
+ for (Map.Entry 刻意不直接序列化 {@link GalleryTask}：那是线上协议对象，字段会随协议演进；日志需要
+ * 稳定的自有结构，并在读取时重建任务对象，避免协议对象被截断写入后重启解析失败。
+ */
+ public static class JournalTask {
+ public int gid;
+ public String name;
+ public byte status;
+ public int proceeding;
+ /** 该任务当前生效的下发操作 id，供报告携带与幂等判断。 */
+ public String operationId;
+ public long revision;
+
+ public JournalTask() {
+ }
+
+ public static JournalTask of(GalleryTask task, String operationId, long revision) {
+ JournalTask journal = new JournalTask();
+ journal.gid = task.getGid();
+ journal.name = task.getName();
+ journal.status = task.getStatus();
+ journal.proceeding = task.getProceeding();
+ journal.operationId = operationId;
+ journal.revision = revision;
+ return journal;
+ }
+
+ /** 重建协议对象；path 属于运行期状态，重启后由扫描重新确认，不写入日志。 */
+ public GalleryTask toGalleryTask() {
+ GalleryTask task = new GalleryTask();
+ task.setGid(gid);
+ task.setName(name);
+ task.setStatus(status);
+ task.setProceeding(proceeding);
+ return task;
+ }
+
+ JournalTask copy() {
+ JournalTask clone = new JournalTask();
+ clone.gid = gid;
+ clone.name = name;
+ clone.status = status;
+ clone.proceeding = proceeding;
+ clone.operationId = operationId;
+ clone.revision = revision;
+ return clone;
+ }
+ }
+
+ /** 已成功执行的命令回执，按 operationId 去重。 */
+ public static class CommandReceipt {
+ public String operationId;
+ public String action;
+ public int gid;
+ public String name;
+ /** 回放时使用的响应码，与当初回给主站的一致。 */
+ public byte result;
+
+ public CommandReceipt() {
+ }
+
+ public static CommandReceipt of(String operationId, String action, int gid, String name, byte result) {
+ CommandReceipt receipt = new CommandReceipt();
+ receipt.operationId = operationId;
+ receipt.action = action;
+ receipt.gid = gid;
+ receipt.name = name;
+ receipt.result = result;
+ return receipt;
+ }
+
+ CommandReceipt copy() {
+ CommandReceipt clone = new CommandReceipt();
+ clone.operationId = operationId;
+ clone.action = action;
+ clone.gid = gid;
+ clone.name = name;
+ clone.result = result;
+ return clone;
+ }
+ }
+
+ /**
+ * 一份不可变的报告快照。
+ *
+ * 任务字段在创建快照时就拷贝，之后任务继续变化也不会改写已发出的报告内容——否则
+ * 重发时会带出与首次不同的数据，主站的幂等落库会以哪一份为准变得不可确定。
+ */
+ public static class ReportSnapshot {
+ public String reportId;
+ public long revision;
+ /** 快照内容用日志结构承载，每个条目自带 revision，供 ACK 时判断是否为更新版本。 */
+ public List 它保证协议 v2 的两条关键性质：
+ * 能力协商结果由 {@link #negotiate} 决定：在收到 version2 hello 之前一律走旧协议路径，
+ * 旧主站不会看到任何它不认识的消息。
+ */
+@Slf4j
+public class TaskReporter {
+
+ private final TaskJournal journal;
+
+ private final Supplier 只有声明了 version2 + reliable-task-v2 才升级；其余情况保持旧协议，
+ * 未知/过低版本不得导致连接中断。
+ */
+ public void negotiate(ProtocolHelloMessage hello) {
+ if (hello != null && hello.supportsReliableTask()) {
+ if (!reliableV2)
+ log.info("主站已协商可靠任务协议 v2");
+ reliableV2 = true;
+ } else {
+ log.info("对端未声明可靠任务协议，继续使用旧上报路径");
+ }
+ }
+
+ public boolean reliableV2() {
+ return reliableV2;
+ }
+
+ /** 连接断开后复位协商结果；新连接必须重新证明支持 v2。 */
+ public void resetNegotiation() {
+ reliableV2 = false;
+ }
+
+ public boolean hasPendingReports() {
+ return !journal.pendingReports().isEmpty();
+ }
+
+ public Collection 调用方必须在本方法成功返回之后才写通道；返回即代表内容已落盘可重发。
+ */
+ public TaskJournal.ReportSnapshot persistSnapshot(List 它只做两件事：移除该报告的待发快照，并回传快照内容供调用方判断哪些任务可以真正
+ * 从队列移除。判断「能否移除」必须由持有运行期队列的调用方完成——ACK 只说明主站已收到
+ * 报告，并不说明任务已经完成。
+ *
+ * @return 被确认的快照；报告未知、失败或版本已被更新时返回 {@code null}
+ */
+ public TaskJournal.ReportSnapshot settleAck(String reportId, boolean success) throws IOException {
+ TaskJournal.ReportSnapshot snapshot = journal.pendingReport(reportId);
+ if (snapshot == null) {
+ log.debug("收到未知或重复的报告 ACK，忽略: reportId={}", shortId(reportId));
+ return null;
+ }
+ if (!success) {
+ log.warn("主站未提交报告，保留快照等待重试: reportId={}, revision={}", shortId(reportId), snapshot.revision);
+ return null;
+ }
+ journal.removePendingReport(reportId, snapshot.revision);
+ log.info("报告已确认: reportId={}, revision={}", shortId(reportId), snapshot.revision);
+ return snapshot;
+ }
+
+ private static String shortId(String id) {
+ return id == null ? null : id.substring(0, Math.min(8, id.length()));
+ }
+}
diff --git a/src/main/java/lion/storageNode.java b/src/main/java/lion/storageNode.java
index c250774..aa9a479 100644
--- a/src/main/java/lion/storageNode.java
+++ b/src/main/java/lion/storageNode.java
@@ -9,6 +9,9 @@ import lion.Service.SubscriptionSnapshotStore;
import lion.Service.DownloadCheckService;
import lion.Service.PrimaryChannelTracker;
import lion.Service.RekickPolicy;
+import lion.Service.ReliableTaskService;
+import lion.Service.TaskJournal;
+import lion.Service.NodeErrorCode;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
@@ -19,8 +22,11 @@ import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import java.net.Socket;
+import java.nio.file.Path;
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
@@ -61,19 +67,73 @@ public class storageNode {
final ExecutorService subscriptionApplyExecutor;
+ /**
+ * 可靠任务协议的执行器。
+ *
+ * 命令处理与日志落盘都是阻塞式磁盘 I/O，绝不能放在 Netty 事件循环上；单线程还顺带
+ * 保证同一 gid 的「命令 → 回执 → 上报」顺序确定。
+ */
+ final ExecutorService reliableExecutor;
+
+ /** 可靠任务协议的状态编排；为 null 时节点只跑旧协议。 */
+ ReliableTaskService reliableTaskService;
+
+ /**
+ * 每份待 ACK 报告最近一次写入的通道与时刻。
+ *
+ * 判断重发同时看两件事：通道是否变化、是否超过重发间隔。只看通道的话，TCP 仍连着、
+ * 但主站没落库或 ACK 在链路上丢失时通道不会变，报告会永远停在待确认；只看时间又会在
+ * 重连后仍等满一个间隔。写入失败时该记录会被清除，下一轮立即重试。
+ */
+ final Map {@code startNetwork=false} 专供测试：不绑定端口、不主动订阅唤醒主站、不启动调度器，
+ * 因此可以在无网络、无真实主站的条件下直接驱动 {@link #mainThread()} 与消息处理器。
+ * 这避免了历史测试会去连接真实主站端口、或与生产争用固定端口的问题。
+ *
+ * @param journalDirectory 任务日志目录；为 null 时该节点启用旧协议路径
+ */
+ storageNode(SubscriptionSnapshotStore subscriptionSnapshotStore, Path journalDirectory, boolean startNetwork){
this.subscriptionSnapshotStore = subscriptionSnapshotStore;
this.subscriptionApplyExecutor = Executors.newSingleThreadExecutor(r -> {
Thread thread = new Thread(r, "subscription-snapshot-apply");
thread.setDaemon(true);
return thread;
});
+ this.reliableExecutor = Executors.newSingleThreadExecutor(r -> {
+ Thread thread = new Thread(r, "storage-node-reliable");
+ thread.setDaemon(true);
+ return thread;
+ });
queue = new ConcurrentHashMap<>();
tempQueue = new HashMap<>();
lock = new ReentrantLock();
+ if (!startNetwork) {
+ downloadCheckService = new DownloadCheckService(queue, false);
+ if (journalDirectory != null)
+ reliableTaskService = restoreReliableService(journalDirectory);
+ return;
+ }
+
int real_port = CustomUtil._findIdlePort(26321);
ChannelFuture bindFuture = new ServerBootstrap()
@@ -101,10 +161,28 @@ public class storageNode {
wakeMainServer();
downloadCheckService = new DownloadCheckService(queue);
+ if (journalDirectory != null)
+ reliableTaskService = restoreReliableService(journalDirectory);
checkThreadPool = Executors.newScheduledThreadPool(1);
checkThreadPool.scheduleAtFixedRate(this::mainThread, 5, 5, TimeUnit.SECONDS);
}
+ /**
+ * 建立可靠任务服务并恢复上次运行的任务与待 ACK 报告。
+ *
+ * 恢复失败（例如日志损坏）直接抛出并阻止启动：带着一份空日志继续跑，会把主站上已
+ * 完成的任务重新当成未完成，甚至重放已执行的删除副作用。宁可拒绝启动，由运维人工处理。
+ */
+ private ReliableTaskService restoreReliableService(Path journalDirectory){
+ ReliableTaskService service = new ReliableTaskService(journalDirectory, queue, downloadCheckService, Path.of(storagePath));
+ try {
+ service.restore();
+ } catch (java.io.IOException e) {
+ throw new IllegalStateException("恢复任务日志失败，拒绝以不一致状态启动: " + journalDirectory, e);
+ }
+ return service;
+ }
+
/**
* 在独立线程里重新唤起主站。
*
@@ -182,6 +260,9 @@ public class storageNode {
//检查，当任务状态发生变化即方法返回true时，再更新，否则return
if (!downloadCheckService.downloadCheck()) {
+ // 即便本轮没有状态变化，也必须先重试待 ACK 报告：连接可能刚刚恢复，
+ // 而重发不能依赖「有任务变化」这个条件，否则 ACK 丢失后会一直卡住。
+ retryPendingReportsIfNegotiated();
boolean isSkip = true;
//返回false之后，还要额外检查是否有压缩完成的任务
if(!queue.isEmpty())
@@ -206,15 +287,39 @@ public class storageNode {
//没有待上报任务时不必发送空数组（此时 downloadCheck 为 false 且无压缩完成任务）。
if (queue.isEmpty())
return;
+ // 已协商 v2：走可靠报告路径（快照先落盘、任务留待 ACK 后才清理），
+ // 旧路径的「发完即出队」在没有确认的链路上会丢任务。
+ if (reliableTaskService != null && reliableTaskService.reporter().reliableV2()) {
+ flushReport();
+ return;
+ }
//发送
//上锁后再发送，避免出现发送完之后再下载完成
lock.lock();
try {
DownloadStatusMessage downloadStatusMessage = new DownloadStatusMessage();
downloadStatusMessage.setGalleryTasks(queue.values().toArray(GalleryTask[]::new));
- target.writeAndFlush(downloadStatusMessage);
+ // 旧协议没有 ACK，只能在写成功后才允许移除：写失败（对方已断开、缓冲区报错等）
+ // 时如果照样出队，完成状态会永久丢失且再也不会重发。
+ java.util.Set 用于「本轮无状态变化」的路径：任务不再变化并不意味着报告已被确认，连接恢复后
+ * 仍必须把旧快照重发出去。
+ */
+ private void retryPendingReportsIfNegotiated(){
+ if (reliableTaskService == null || !reliableTaskService.reporter().reliableV2())
+ return;
+ if (reliableTaskService.pendingReports().isEmpty())
+ return;
+ Channel target = primaryChannel.current();
+ if (target == null)
+ return;
+ long now = System.currentTimeMillis();
+ for (TaskJournal.ReportSnapshot pending : reliableTaskService.pendingReports()) {
+ ReportSend previous = reportSends.get(pending.reportId);
+ boolean sameChannel = previous != null && previous.channel() == target;
+ boolean fresh = previous != null && now - previous.sentAt() < reportRetryIntervalMillis;
+ if (sameChannel && fresh)
+ continue;
+ sendReport(target, pending);
+ log.info("重试待确认报告: reportId={}, revision={}", pending.reportId, pending.revision);
+ }
+ }
+
/**
* 该消息是否只有主站会发。
*
@@ -241,15 +372,106 @@ public class storageNode {
IdentityMessage identity = (IdentityMessage) message;
return "lionwebsite".equals(identity.getIdentity());
}
+ // v2 能力协商只由主站发起，同样构成主站证据；即便对端版本过低也先认领通道。
+ if (message.messageType == AbstractMessage.PROTOCOL_HELLO_MESSAGE)
+ return true;
return message.messageType == AbstractMessage.DOWNLOAD_POST_MESSAGE
|| message.messageType == AbstractMessage.DELETE_GALLERY_MESSAGE
|| message.messageType == AbstractMessage.AVAILABLE_CHECK_MESSAGE
|| message.messageType == AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE;
}
+ /**
+ * 在可靠执行器上处理命令，并在完成后立即把结果回给主站。
+ *
+ * 结果必须回到当初收到命令的那条 channel：主站按 messageId 与通道配对等待响应，
+ * 若改写到「当前主站通道」上，命令与响应可能落在不同连接而永远配不上。
+ */
+ private void handleTaskCommandAsync(ChannelHandlerContext ctx, TaskCommandMessage command){
+ reliableExecutor.execute(() -> {
+ byte result;
+ try {
+ result = reliableTaskService.handleCommand(command);
+ } catch (Exception e) {
+ log.error("处理可靠任务命令失败: operationId={}", command.getOperationId(), e);
+ result = NodeErrorCode.IO_ERROR;
+ }
+ // 命令成功后立刻尝试上报一份新状态，让主站尽快看到结果。
+ if (NodeErrorCode.isTerminal(result))
+ flushReportAsync();
+ ctx.writeAndFlush(new ResponseMessage(command.messageId, result));
+ });
+ }
+
+ /**
+ * 生成并发送一份可靠报告（若已协商 v2）。
+ *
+ * 顺序至关重要：先落盘快照，再写通道。写失败时快照仍留在日志里，重连后重发；
+ * 因此任何一次发送失败都不会让任务从节点侧消失。
+ */
+ void flushReportAsync(){
+ reliableExecutor.execute(this::flushReport);
+ }
+
+ /** 同步版本，供调度线程与测试直接调用。 */
+ void flushReport(){
+ if (reliableTaskService == null || !reliableTaskService.reporter().reliableV2())
+ return;
+ try {
+ // 先把扫描出的最新进度对齐日志，保证报告内容与磁盘事实一致。
+ reliableTaskService.refreshJournalFromQueue();
+ Channel target = primaryChannel.current();
+ if (target == null)
+ return;
+
+ // 断线后重发尚未 ACK 的快照，且只在「换了通道」时发一次，避免每轮调度重复轰炸。
+ for (TaskJournal.ReportSnapshot pending : reliableTaskService.pendingReports()) {
+ ReportSend previous = reportSends.get(pending.reportId);
+ // 未发过、通道已变更、或超过重发间隔：都需要重发（reportId 与 revision 保持不变）。
+ boolean sameChannel = previous != null && previous.channel() == target;
+ boolean fresh = previous != null
+ && System.currentTimeMillis() - previous.sentAt() < reportRetryIntervalMillis;
+ if (sameChannel && fresh)
+ continue;
+ sendReport(target, pending);
+ log.info("重发待确认报告: reportId={}, revision={}", pending.reportId, pending.revision);
+ }
+
+ Optional 只有真正写成功才记时刻：若先记后写，一次写失败会被当成「刚发过」而白白等一个
+ * 重发周期。监听器在写失败时也会清除登记，覆盖「已入队但最终写失败」的情况。
+ */
+ private void sendReport(Channel target, TaskJournal.ReportSnapshot snapshot){
+ String reportId = snapshot.reportId;
+ ChannelFuture future = target.writeAndFlush(reliableTaskService.reporter().toMessage(snapshot));
+ reportSends.put(reportId, new ReportSend(target, System.currentTimeMillis()));
+ future.addListener(result -> {
+ if (!result.isSuccess()) {
+ reportSends.remove(reportId);
+ log.warn("报告写入通道失败，下一轮重发: reportId={}", reportId);
+ }
+ });
+ }
+
class MyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{
Map 线程都是守护线程，生产进程退出不依赖本方法；但测试与热重启需要显式释放，
+ * 否则会残留仍在调度或写日志的线程。
+ */
+ public void shutdown(){
+ if (checkThreadPool != null)
+ checkThreadPool.shutdownNow();
+ rekickExecutor.shutdownNow();
+ subscriptionApplyExecutor.shutdownNow();
+ reliableExecutor.shutdownNow();
+ if (downloadCheckService != null)
+ downloadCheckService.shutdown();
+ }
}
diff --git a/src/main/resources/config.properties b/src/main/resources/config.properties
index 94f22ad..011f441 100644
--- a/src/main/resources/config.properties
+++ b/src/main/resources/config.properties
@@ -7,3 +7,5 @@ SubscriptionMaxPayloadBytes=52428800
SubscriptionHttpPort=8889
SubscriptionHttpWorkers=4
SubscriptionSocketTimeoutMs=10000
+# 可靠任务协议的任务日志目录（独立于订阅数据目录，默认不接触订阅/测试生产目录）
+TaskJournalDir=/root/gallery/storageNode/task-journal
diff --git a/src/main/resources/reflect-config.json b/src/main/resources/reflect-config.json
index ccb89e3..fa39818 100644
--- a/src/main/resources/reflect-config.json
+++ b/src/main/resources/reflect-config.json
@@ -107,6 +107,78 @@
"allDeclaredFields" : true,
"allPublicFields" : true
},
+ {
+ "name": "lion.Message.Main.ProtocolHelloMessage",
+ "allDeclaredConstructors" : true,
+ "allPublicConstructors" : true,
+ "allDeclaredMethods" : true,
+ "allPublicMethods" : true,
+ "allDeclaredFields" : true,
+ "allPublicFields" : true
+ },
+ {
+ "name": "lion.Message.Main.TaskCommandMessage",
+ "allDeclaredConstructors" : true,
+ "allPublicConstructors" : true,
+ "allDeclaredMethods" : true,
+ "allPublicMethods" : true,
+ "allDeclaredFields" : true,
+ "allPublicFields" : true
+ },
+ {
+ "name": "lion.Message.Main.TaskReportMessage",
+ "allDeclaredConstructors" : true,
+ "allPublicConstructors" : true,
+ "allDeclaredMethods" : true,
+ "allPublicMethods" : true,
+ "allDeclaredFields" : true,
+ "allPublicFields" : true
+ },
+ {
+ "name": "lion.Message.Main.TaskReportAckMessage",
+ "allDeclaredConstructors" : true,
+ "allPublicConstructors" : true,
+ "allDeclaredMethods" : true,
+ "allPublicMethods" : true,
+ "allDeclaredFields" : true,
+ "allPublicFields" : true
+ },
+ {
+ "name": "lion.Service.TaskJournal$State",
+ "allDeclaredConstructors" : true,
+ "allPublicConstructors" : true,
+ "allDeclaredMethods" : true,
+ "allPublicMethods" : true,
+ "allDeclaredFields" : true,
+ "allPublicFields" : true
+ },
+ {
+ "name": "lion.Service.TaskJournal$JournalTask",
+ "allDeclaredConstructors" : true,
+ "allPublicConstructors" : true,
+ "allDeclaredMethods" : true,
+ "allPublicMethods" : true,
+ "allDeclaredFields" : true,
+ "allPublicFields" : true
+ },
+ {
+ "name": "lion.Service.TaskJournal$CommandReceipt",
+ "allDeclaredConstructors" : true,
+ "allPublicConstructors" : true,
+ "allDeclaredMethods" : true,
+ "allPublicMethods" : true,
+ "allDeclaredFields" : true,
+ "allPublicFields" : true
+ },
+ {
+ "name": "lion.Service.TaskJournal$ReportSnapshot",
+ "allDeclaredConstructors" : true,
+ "allPublicConstructors" : true,
+ "allDeclaredMethods" : true,
+ "allPublicMethods" : true,
+ "allDeclaredFields" : true,
+ "allPublicFields" : true
+ },
{
"name": "lion.Message.AbstractMessage",
"allDeclaredConstructors" : true,
diff --git a/src/test/java/lion/CustomUtilTest.java b/src/test/java/lion/CustomUtilTest.java
index d59cfbc..e9be0e9 100644
--- a/src/test/java/lion/CustomUtilTest.java
+++ b/src/test/java/lion/CustomUtilTest.java
@@ -2,33 +2,19 @@ package lion;
import org.junit.jupiter.api.Test;
-import java.io.BufferedReader;
-import java.io.StringReader;
import java.net.ServerSocket;
-import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
-/** 下载链路用到的两个工具方法：断点续传请求头解析与空闲端口探测。 */
+/**
+ * CustomUtil 保留的工具方法：空闲端口探测。
+ *
+ * 原先针对 {@code getRequestHeader} 的用例已随 Range 解析迁移到
+ * {@code lion.Http.ByteRangeTest}，那里覆盖普通、后缀、开放结尾与非法区间。
+ */
class CustomUtilTest {
- /** Range 头必须原样取出「bytes=...」，供 sendFileRange 计算区间。 */
- @Test
- void extractsRangeHeader() throws Exception {
- BufferedReader reader = new BufferedReader(new StringReader(
- "Host: lionwebsite.xyz\r\nUser-Agent: Mozilla/5.0\r\nRange: bytes=1024-2047\r\n\r\n"));
-
- assertEquals("bytes=1024-2047", CustomUtil.getRequestHeader(reader));
- }
-
- /** 没有 Range 头（全新下载）时返回 null，调用方据此从头开始传。 */
- @Test
- void returnsNullWhenRangeHeaderIsAbsent() throws Exception {
- BufferedReader reader = new BufferedReader(new StringReader(
- "Host: lionwebsite.xyz\r\nUser-Agent: Mozilla/5.0\r\n\r\n"));
-
- assertNull(CustomUtil.getRequestHeader(reader));
- }
-
/** 从指定端口开始探测，且返回的端口确实可以立刻绑定。 */
@Test
void returnsBindableIdlePort() throws Exception {
diff --git a/src/test/java/lion/Externel/BackupSubServerTest.java b/src/test/java/lion/Externel/BackupSubServerTest.java
index 1038e27..817c60f 100644
--- a/src/test/java/lion/Externel/BackupSubServerTest.java
+++ b/src/test/java/lion/Externel/BackupSubServerTest.java
@@ -50,16 +50,20 @@ class BackupSubServerTest {
static int readyPort;
static int emptyPort;
+ static BackupSubServer readyServer;
+ static BackupSubServer emptyServer;
@BeforeAll
static void startServers() throws Exception {
// 有有效快照的备机：正常分发。
SubscriptionSnapshotStore ready = store(tempDir.resolve("ready"));
assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, ready.apply(snapshot()).code());
- readyPort = startServer(ready);
+ readyServer = new BackupSubServer(ready, 0, 2);
+ readyPort = readyServer.startServer().port();
// 从未收到过快照的备机：必须回 503，而不是把请求静默挂起。
- emptyPort = startServer(store(tempDir.resolve("empty")));
+ emptyServer = new BackupSubServer(store(tempDir.resolve("empty")), 0, 2);
+ emptyPort = emptyServer.startServer().port();
}
/** 命中公开 Key：v2 与 clash 各自返回对应正文。 */
@@ -140,23 +144,53 @@ class BackupSubServerTest {
assertEquals(416, request(readyPort, "GET", "/sub/v2/" + PUBLIC_KEY, "Range: bytes=999-1000").status());
}
+ /** 后缀区间 bytes=-4 取正文末尾（旧实现不支持后缀写法）。 */
+ @Test
+ void suffixRangeReturnsTrailingBytes() throws Exception {
+ String expected = V2_CONTENT.substring(V2_CONTENT.length() - 4);
+ Response response = request(readyPort, "GET", "/sub/v2/" + PUBLIC_KEY, "Range: bytes=-4");
+
+ assertEquals(206, response.status());
+ assertEquals(expected, response.bodyAsString());
+ assertEquals("bytes " + (V2_CONTENT.length() - 4) + "-" + (V2_CONTENT.length() - 1) + "/" + V2_CONTENT.length(),
+ response.header("content-range"));
+ }
+
+ /** 开放结尾 bytes=4- 一直取到正文末尾。 */
+ @Test
+ void openEndedRangeGoesToEnd() throws Exception {
+ Response response = request(readyPort, "GET", "/sub/v2/" + PUBLIC_KEY, "Range: bytes=4-");
+
+ assertEquals(206, response.status());
+ assertEquals(V2_CONTENT.substring(4), response.bodyAsString());
+ }
+
+ /** 关闭服务后端口必须释放，且可重复关闭。 */
+ @Test
+ void serverShutdownReleasesPort(@TempDir Path temp) throws Exception {
+ SubscriptionSnapshotStore store = store(temp.resolve("shutdown"));
+ assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(snapshot()).code());
+
+ BackupSubServer server = new BackupSubServer(store, 0, 2);
+ int port = server.startServer().port();
+
+ // 先确认真的在服务。
+ assertEquals(200, get(port, "/sub/v2/" + PUBLIC_KEY).status());
+
+ server.server().close();
+ server.server().close(); // 重复关闭不得抛异常
+
+ try (ServerSocket rebind = new ServerSocket(port)) {
+ assertEquals(port, rebind.getLocalPort(), "关闭后端口必须可重新绑定");
+ }
+ }
+
// ---- 测试脚手架 ----
private static SubscriptionSnapshotStore store(Path root) {
return new SubscriptionSnapshotStore(root, SECRET, 3600, 1024 * 1024);
}
- private static int startServer(SubscriptionSnapshotStore store) throws IOException {
- int port;
- try (ServerSocket probe = new ServerSocket(0)) {
- port = probe.getLocalPort();
- }
- Thread thread = new Thread(new BackupSubServer(store, port, 2), "backup-sub-test-" + port);
- thread.setDaemon(true);
- thread.start();
- return port;
- }
-
private static Response get(int port, String path) throws IOException {
return request(port, "GET", path);
}
diff --git a/src/test/java/lion/Http/ByteRangeTest.java b/src/test/java/lion/Http/ByteRangeTest.java
new file mode 100644
index 0000000..106b396
--- /dev/null
+++ b/src/test/java/lion/Http/ByteRangeTest.java
@@ -0,0 +1,107 @@
+package lion.Http;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/** Range 头解析的边界矩阵：普通、后缀、开放结尾、非法与不可满足。 */
+class ByteRangeTest {
+
+ @Test
+ void absentOrUnsupportedHeaderMeansFullEntity() {
+ assertEquals(ByteRange.Kind.FULL, ByteRange.parse(null, 10).kind());
+ assertEquals(ByteRange.Kind.FULL, ByteRange.parse("items=0-3", 10).kind());
+ assertEquals(ByteRange.Kind.FULL, ByteRange.parse("bytes=", 10).kind());
+ assertEquals(ByteRange.Kind.FULL, ByteRange.parse("bytes=abc-def", 10).kind());
+ }
+
+ @Test
+ void plainRangeIsPartial() {
+ ByteRange.Resolution range = ByteRange.parse("bytes=2-5", 10);
+ assertEquals(ByteRange.Kind.PARTIAL, range.kind());
+ assertEquals(2, range.start());
+ assertEquals(5, range.endInclusive());
+ assertEquals(4, range.length());
+ assertEquals("bytes 2-5/10", range.contentRange(10));
+ }
+
+ @Test
+ void endBeyondFileIsClamped() {
+ ByteRange.Resolution range = ByteRange.parse("bytes=8-999", 10);
+ assertEquals(8, range.start());
+ assertEquals(9, range.endInclusive());
+ }
+
+ @Test
+ void openEndedRangeCoversToFileEnd() {
+ ByteRange.Resolution range = ByteRange.parse("bytes=7-", 10);
+ assertTrue(range.partial());
+ assertEquals(7, range.start());
+ assertEquals(9, range.endInclusive());
+ }
+
+ @Test
+ void suffixRangeTakesLastBytes() {
+ ByteRange.Resolution range = ByteRange.parse("bytes=-3", 10);
+ assertTrue(range.partial());
+ assertEquals(7, range.start());
+ assertEquals(9, range.endInclusive());
+ }
+
+ /** 后缀长度超过实体本身时返回整个实体，而不是负数起点。 */
+ @Test
+ void suffixLongerThanFileCoversWholeEntity() {
+ ByteRange.Resolution range = ByteRange.parse("bytes=-100", 10);
+ assertEquals(0, range.start());
+ assertEquals(9, range.endInclusive());
+ }
+
+ /** 后缀长度为 0 是合法语法但不可满足。 */
+ @Test
+ void zeroLengthSuffixIsUnsatisfiable() {
+ assertTrue(ByteRange.parse("bytes=-0", 10).unsatisfiable());
+ }
+
+ @Test
+ void startBeyondFileIsUnsatisfiable() {
+ assertTrue(ByteRange.parse("bytes=10-", 10).unsatisfiable());
+ assertTrue(ByteRange.parse("bytes=999-1000", 10).unsatisfiable());
+ }
+
+ /** 起点晚于终点属于不可满足。 */
+ @Test
+ void reversedRangeIsUnsatisfiable() {
+ assertTrue(ByteRange.parse("bytes=6-2", 10).unsatisfiable());
+ }
+
+ /** 空文件上的任何区间都不可满足。 */
+ @Test
+ void emptyFileHasNoSatisfiableRange() {
+ assertTrue(ByteRange.parse("bytes=0-0", 0).unsatisfiable());
+ assertTrue(ByteRange.parse("bytes=-1", 0).unsatisfiable());
+ }
+
+ /** 多区间只取第一段，与旧实现一致。 */
+ @Test
+ void multipleRangesUseFirstSegmentOnly() {
+ ByteRange.Resolution range = ByteRange.parse("bytes=1-2,5-6", 10);
+ assertEquals(1, range.start());
+ assertEquals(2, range.endInclusive());
+ }
+
+ /** 单位大小写不敏感。 */
+ @Test
+ void rangeUnitIsCaseInsensitive() {
+ assertTrue(ByteRange.parse("Bytes=1-2", 10).partial());
+ }
+
+ /** 越界的 Long 值不得抛异常，退回完整实体。 */
+ @Test
+ void overflowingNumbersFallBackToFullEntity() {
+ ByteRange.Resolution range = ByteRange.parse("bytes=99999999999999999999-", 10);
+ assertEquals(ByteRange.Kind.FULL, range.kind());
+ assertFalse(range.partial());
+ }
+}
diff --git a/src/test/java/lion/Http/GalleryArchiveIndexTest.java b/src/test/java/lion/Http/GalleryArchiveIndexTest.java
new file mode 100644
index 0000000..9abdd3e
--- /dev/null
+++ b/src/test/java/lion/Http/GalleryArchiveIndexTest.java
@@ -0,0 +1,137 @@
+package lion.Http;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/** gid → 归档索引：命中、重扫发现新归档、失效与标记提取。 */
+class GalleryArchiveIndexTest {
+
+ @Test
+ void scansDirectoryOnceThenServesFromIndex(@TempDir Path root) throws IOException {
+ writeArchive(root, "Title [123]", "zip");
+ AtomicInteger scans = new AtomicInteger();
+ GalleryArchiveIndex index = new GalleryArchiveIndex(10_000L, System::currentTimeMillis,
+ counting(root, scans));
+
+ assertNotNull(index.find("123"));
+ assertNotNull(index.find("123"));
+ assertEquals(1, scans.get(), "命中不应重复扫描目录");
+ }
+
+ /** 归档在索引建立后出现，超过重扫间隔必须能被发现。 */
+ @Test
+ void newlyAddedArchiveBecomesVisibleAfterRescanWindow(@TempDir Path root) throws IOException {
+ AtomicLong now = new AtomicLong(1_000L);
+ GalleryArchiveIndex index = new GalleryArchiveIndex(5_000L, now::get,
+ () -> root.toFile().listFiles(File::isDirectory));
+
+ assertNull(index.find("555"));
+
+ writeArchive(root, "Later [555]", "zip");
+ now.addAndGet(5_000L);
+
+ assertNotNull(index.find("555"), "超过重扫间隔后新归档必须可见");
+ }
+
+ /** 未超过重扫间隔时不得反复扫描同一批无意义 gid。 */
+ @Test
+ void repeatedMissesWithinWindowDoNotRescan(@TempDir Path root) throws IOException {
+ writeArchive(root, "Title [123]", "zip");
+ AtomicInteger scans = new AtomicInteger();
+ GalleryArchiveIndex index = new GalleryArchiveIndex(10_000L, System::currentTimeMillis,
+ counting(root, scans));
+
+ assertNull(index.find("999"));
+ assertNull(index.find("998"));
+ assertEquals(1, scans.get(), "间隔内不应重复扫描");
+ }
+
+ @Test
+ void invalidateForcesFreshScan(@TempDir Path root) throws IOException {
+ AtomicInteger scans = new AtomicInteger();
+ GalleryArchiveIndex index = new GalleryArchiveIndex(10_000L, System::currentTimeMillis,
+ counting(root, scans));
+ assertNull(index.find("42"));
+
+ writeArchive(root, "Fresh [42]", "zip");
+ index.invalidate();
+
+ assertNotNull(index.find("42"), "invalidate 后必须重新扫描");
+ assertEquals(2, scans.get());
+ }
+
+ /** 索引只收录归档真实存在的目录。 */
+ @Test
+ void directoryWithoutArchiveIsNotIndexed(@TempDir Path root) throws IOException {
+ Files.createDirectories(root.resolve("No archive [900]"));
+ GalleryArchiveIndex index = new GalleryArchiveIndex(10_000L, System::currentTimeMillis,
+ () -> root.toFile().listFiles(File::isDirectory));
+
+ assertNull(index.find("900"));
+ assertEquals(0, index.size());
+ }
+
+ /** 非数字、空白与 null 的 gid 直接返回 null，不触碰磁盘。 */
+ @Test
+ void invalidGidNeverHitsDisk(@TempDir Path root) {
+ AtomicInteger scans = new AtomicInteger();
+ GalleryArchiveIndex index = new GalleryArchiveIndex(10_000L, System::currentTimeMillis,
+ counting(root, scans));
+
+ assertNull(index.find(null));
+ assertNull(index.find(""));
+ assertNull(index.find(" "));
+ assertNull(index.find("abc"));
+ assertEquals(0, scans.get());
+ }
+
+ /** 目录名里的括号数字按「标记后紧跟 ] 或 -」提取，忽略不合规的方括号。 */
+ @Test
+ void extractsOnlyWellFormedGidMarkers() {
+ assertEquals(List.of(123), GalleryArchiveIndex.extractGids("Title [123]"));
+ assertEquals(List.of(123), GalleryArchiveIndex.extractGids("Title [123]-extra"));
+ assertEquals(List.of(7, 42), GalleryArchiveIndex.extractGids("A [7] B [42]"));
+ assertTrue(GalleryArchiveIndex.extractGids("Title [abc]").isEmpty());
+ assertTrue(GalleryArchiveIndex.extractGids("Title [123").isEmpty(), "缺少闭合符不算标记");
+ assertTrue(GalleryArchiveIndex.extractGids("Title 123]").isEmpty());
+ assertTrue(GalleryArchiveIndex.extractGids("").isEmpty());
+ assertTrue(GalleryArchiveIndex.extractGids(null).isEmpty());
+ }
+
+ /** 同一个 gid 在多个目录出现时，命中其中之一（与逐请求扫描同类）。 */
+ @Test
+ void duplicateGidResolvesToAnExistingArchive(@TempDir Path root) throws IOException {
+ writeArchive(root, "A [77]", "first");
+ writeArchive(root, "B [77]", "second");
+ GalleryArchiveIndex index = new GalleryArchiveIndex(10_000L, System::currentTimeMillis,
+ () -> root.toFile().listFiles(File::isDirectory));
+
+ File found = index.find("77");
+ assertNotNull(found);
+ assertTrue(found.isFile());
+ assertTrue(found.getName().endsWith(".zip"));
+ }
+
+ private static java.util.function.Supplier 处理器只返回固定字节，不需要文件系统；端口用 0 由系统分配。
+ */
+class HttpFileServerTest {
+
+ /** 执行器必须是有界队列 + CallerRuns，不能是无上限堆积的队列。 */
+ @Test
+ void executorUsesBoundedQueueWithCallerRunsBackpressure(@TempDir Path ignored) throws Exception {
+ HttpFileServer server = HttpFileServer.create(0, "/",
+ request -> HttpResponseModel.text(200, "text/plain", "ok"), 2, 3);
+ try {
+ ThreadPoolExecutor pool = assertInstanceOf(ThreadPoolExecutor.class, server.executor());
+ var queue = assertInstanceOf(ArrayBlockingQueue.class, pool.getQueue());
+ assertEquals(3, queue.remainingCapacity(), "队列深度必须是显式上限");
+ assertEquals(2, pool.getMaximumPoolSize());
+ assertInstanceOf(ThreadPoolExecutor.CallerRunsPolicy.class, pool.getRejectedExecutionHandler());
+ } finally {
+ server.close();
+ }
+ }
+
+ /** 关闭后端口必须立即释放，且可重复关闭。 */
+ @Test
+ void closeReleasesPortAndIsIdempotent(@TempDir Path ignored) throws Exception {
+ HttpFileServer server = HttpFileServer.create(0, "/",
+ request -> HttpResponseModel.text(200, "text/plain", "ok"), 2);
+ int port = server.port();
+ server.start();
+
+ assertEquals(200, status(port, "GET", "/"));
+
+ server.close();
+ server.close(); // 重复关闭不得抛异常
+
+ try (ServerSocket rebind = new ServerSocket(port)) {
+ assertEquals(port, rebind.getLocalPort(), "关闭后端口必须可重新绑定");
+ }
+ }
+
+ /** 未启动就关闭不得抛异常。 */
+ @Test
+ void closeBeforeStartIsSafe(@TempDir Path ignored) throws Exception {
+ HttpFileServer server = HttpFileServer.create(0, "/",
+ request -> HttpResponseModel.text(200, "text/plain", "ok"), 1);
+ assertDoesNotThrow(server::close);
+ }
+
+ /** 处理器抛异常时必须回 500，且外壳继续可用。 */
+ @Test
+ void handlerFailureBecomesInternalServerError(@TempDir Path ignored) throws Exception {
+ HttpFileServer server = HttpFileServer.create(0, "/", request -> {
+ throw new IllegalStateException("boom");
+ }, 2);
+ try {
+ server.start();
+ assertEquals(500, status(server.port(), "GET", "/"));
+ } finally {
+ server.close();
+ }
+ }
+
+ /** 请求模型暴露的路径保留字面量 +，查询值按表单语义把 + 解码为空格。 */
+ @Test
+ void requestModelDecodesPathAndQueryConsistently(@TempDir Path ignored) throws Exception {
+ HttpFileServer server = HttpFileServer.create(0, "/",
+ request -> HttpResponseModel.text(200, "text/plain",
+ (request.decodedPath() == null ? "" : request.decodedPath())
+ + "|" + request.parameter("AuthCode")),
+ 2);
+ try {
+ server.start();
+ assertEquals("/plus+name.zip|a b", body(server.port(), "GET", "/plus+name.zip?AuthCode=a+b"));
+ } finally {
+ server.close();
+ }
+ }
+
+ private static int status(int port, String method, String path) throws IOException {
+ return (int) parse(port, method, path)[0];
+ }
+
+ private static String body(int port, String method, String path) throws IOException {
+ return new String((byte[]) parse(port, method, path)[1], StandardCharsets.UTF_8);
+ }
+
+ /** 返回 [status, bodyBytes]。 */
+ private static Object[] parse(int port, String method, String path) throws IOException {
+ try (Socket socket = new Socket("127.0.0.1", port)) {
+ socket.setSoTimeout(5_000);
+ String request = method + " " + path + " HTTP/1.1" + CRLF
+ + "Host: 127.0.0.1" + CRLF + "Connection: close" + CRLF + CRLF;
+ socket.getOutputStream().write(request.getBytes(StandardCharsets.ISO_8859_1));
+ socket.getOutputStream().flush();
+
+ byte[] raw = socket.getInputStream().readAllBytes();
+ int split = indexOfHeaderEnd(raw);
+ assertTrue(split > 0, "响应必须含完整的头部结束标记");
+ String head = new String(raw, 0, split, StandardCharsets.ISO_8859_1);
+ int code = Integer.parseInt(head.split(CRLF)[0].split(" ")[1]);
+ return new Object[]{code, Arrays.copyOfRange(raw, split + 4, raw.length)};
+ }
+ }
+
+ private static int indexOfHeaderEnd(byte[] raw) {
+ for (int i = 0; i + 3 < raw.length; i++)
+ if (raw[i] == 13 && raw[i + 1] == 10 && raw[i + 2] == 13 && raw[i + 3] == 10)
+ return i;
+ return -1;
+ }
+
+ private static final String CRLF = "\r\n";
+}
+
diff --git a/src/test/java/lion/MessageCodecTest.java b/src/test/java/lion/MessageCodecTest.java
index 94e6428..a08eed9 100644
--- a/src/test/java/lion/MessageCodecTest.java
+++ b/src/test/java/lion/MessageCodecTest.java
@@ -13,8 +13,12 @@ import lion.Message.Main.DownloadPostMessage;
import lion.Message.Main.DownloadStatusMessage;
import lion.Message.Main.IdentityMessage;
import lion.Message.Main.MaintainMessage;
+import lion.Message.Main.ProtocolHelloMessage;
import lion.Message.Main.ResponseMessage;
import lion.Message.Main.SubscriptionSnapshotMessage;
+import lion.Message.Main.TaskCommandMessage;
+import lion.Message.Main.TaskReportAckMessage;
+import lion.Message.Main.TaskReportMessage;
import org.junit.jupiter.api.Test;
import java.nio.charset.StandardCharsets;
@@ -40,6 +44,76 @@ class MessageCodecTest {
assertRoundTrip(new MaintainMessage(), AbstractMessage.MAINTAIN_MESSAGE);
assertRoundTrip(new AvailableCheckMessage(), AbstractMessage.AVAILABLE_CHECK_MESSAGE);
assertRoundTrip(snapshot(), AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE);
+ assertRoundTrip(ProtocolHelloMessage.v2(), AbstractMessage.PROTOCOL_HELLO_MESSAGE);
+ assertRoundTrip(taskCommand(), AbstractMessage.TASK_COMMAND_MESSAGE);
+ assertRoundTrip(taskReport(), AbstractMessage.TASK_REPORT_MESSAGE);
+ assertRoundTrip(ack(), AbstractMessage.TASK_REPORT_ACK_MESSAGE);
+ }
+
+ /**
+ * 新增消息的线上字段集必须精确可控。
+ *
+ * 主站使用更严格的 Jackson 并默认拒绝未知字段，任何多出来的属性都会让对端反序列化失败。
+ * Jackson 2 会把 {@code isXxx()} 当成布尔属性序列化，因此 {@code TaskCommandMessage} 的
+ * {@code isDownload()/isDelete()} 与 {@code TaskReportAckMessage.isSuccess()} 必须带
+ * {@code @JsonIgnore}；本用例把实际字段集钉死，防止它们悄悄回到线路上。
+ */
+ @Test
+ void newMessageWireFieldsAreExactlyAsContract() throws Exception {
+ assertEquals(java.util.Set.of("messageType", "messageId", "operationId", "action", "gid", "name"),
+ fieldsOf(taskCommand()), "任务命令不得额外序列化 download/delete 等派生属性");
+ assertEquals(java.util.Set.of("messageType", "messageId", "reportId", "result"),
+ fieldsOf(ack()), "报告确认不得额外序列化 success");
+ assertEquals(java.util.Set.of("messageType", "messageId", "protocolVersion", "features"),
+ fieldsOf(ProtocolHelloMessage.v2()), "hello 不得额外序列化 supportsReliableTask");
+ assertEquals(java.util.Set.of("messageType", "messageId", "reportId", "revision", "galleryTasks", "operationIds"),
+ fieldsOf(taskReport()));
+ // 旧类型绝不能因为基类改动而多出字段：旧主站的 Jackson 会拒绝未知字段。
+ assertEquals(java.util.Set.of("messageType", "messageId", "galleryTasks"),
+ fieldsOf(downloadStatus()), "旧状态消息的字段集不得变化");
+ assertEquals(java.util.Set.of("messageType", "messageId", "identity"),
+ fieldsOf(identity()), "旧身份消息的字段集不得变化");
+ }
+
+ /** 序列化一份消息并取回其顶层 JSON 字段名集合。 */
+ private static java.util.Set 集成用例起在临时端口、只使用 {@link TempDir} 里的临时目录，不读取生产配置、
+ * 不启动 {@code main}。
*/
class MultiThreadedHTTPServerTest {
- /** 普通请求：path 与各查询参数都要被拆出来。 */
+ // ---- 请求行解析（公开兼容入口） ----
+
@Test
void parsesPathAndQueryParameters() {
Map 原实现直接取 {@code requestParts[1]}，空行会抛 ArrayIndexOutOfBoundsException；
- * 它不是 IOException，会被线程池静默吞掉，socket 也关不掉。现在应回 400 并关闭连接。
- */
+ /** 路径里的 {@code +} 是字面量，不得被当成空格（query 值才按 + → 空格）。 */
@Test
- void malformedRequestLineIsRejectedWithoutCrashing() throws Exception {
- try (ServerSocket server = new ServerSocket(0); Socket client = new Socket("127.0.0.1", server.getLocalPort())) {
- client.setSoTimeout(5_000);
- Thread handler = startHandler(server);
+ void keepsPlusLiteralInPathButDecodesItInQuery() {
+ Map 原实现按未夹取的区间写 Content-Length，客户端拿到一个远大于实际发送量的长度，
- * 会一直等下去。现在 Content-Range / Content-Length / 正文都必须只覆盖到文件末尾。
- */
- @Test
- void oversizedRangeIsClampedToFileEnd(@TempDir Path root) throws Exception {
- byte[] payload = "0123456789".getBytes(StandardCharsets.UTF_8);
- Path archive = root.resolve("archive.zip");
- Files.write(archive, payload);
-
- String head = fetchRange(root, archive, "bytes=4-999999");
-
- assertTrue(head.contains("Content-Range: bytes 4-9/10"), "Content-Range 必须夹到文件末尾：" + head);
- assertTrue(head.contains("Content-Length: 6"), "Content-Length 必须与实际发送量一致：" + head);
- }
-
- /** Range 起点超过文件长度时无可发送内容，应回 416，而不是假装有长度。 */
- @Test
- void rangeStartingBeyondFileIsNotSatisfiable(@TempDir Path root) throws Exception {
- Path archive = root.resolve("archive.zip");
- Files.write(archive, "0123456789".getBytes(StandardCharsets.UTF_8));
-
- String head = fetchRange(root, archive, "bytes=999-1000");
-
- assertTrue(head.startsWith("HTTP/1.1 416"), "越界起点应回 416：" + head.lines().findFirst().orElse(""));
+ void pathDecoderPreservesPlusWhileQueryDecoderDoesNot() {
+ assertEquals("/a+b.zip", lion.Http.HttpFileServer.decodePath("/a+b.zip"));
+ assertEquals("/a+b.zip", lion.Http.HttpFileServer.decodePath("/a%2Bb.zip"),
+ "百分号编码的 '+' 解码后同样是 '+'");
+ assertEquals("a b", lion.Http.HttpFileServer.decodeQuery("a+b"));
}
// ---- 管理员直取文件的口令与目录约束 ----
- /** 未配置口令时，管理员能力必须关闭，任何 AuthCode 都不能直取文件。 */
@Test
void adminCapabilityIsOffUntilCodeConfigured() {
lion.Config.Config.adminDownloadCode = "";
@@ -149,7 +98,6 @@ class MultiThreadedHTTPServerTest {
assertFalse(MultiThreadedHTTPServer.isAdminRequest(""));
}
- /** 配置口令后只认该口令，旧写死值与其他值都不再放行。 */
@Test
void onlyConfiguredAdminCodeIsAccepted() {
lion.Config.Config.adminDownloadCode = "s3cret-code";
@@ -163,7 +111,6 @@ class MultiThreadedHTTPServerTest {
}
}
- /** 管理员直取文件只允许根目录之下的真实文件。 */
@Test
void adminDownloadIsConfinedToConfiguredRoot(@TempDir Path root) throws Exception {
Path galleryRoot = Files.createDirectories(root.resolve("gallery"));
@@ -183,7 +130,6 @@ class MultiThreadedHTTPServerTest {
}
}
- /** 指向根目录之外的符号链接也不得把文件泄漏出去。 */
@Test
void adminDownloadRejectsSymlinkEscape(@TempDir Path root) throws Exception {
Path galleryRoot = Files.createDirectories(root.resolve("gallery"));
@@ -202,54 +148,314 @@ class MultiThreadedHTTPServerTest {
}
}
+ // ---- 真实 HTTP 集成（临时端口 + 临时目录） ----
+
+ /**
+ * 管理员直取：路径里含字面量 '+' 的文件名必须能按实际路径打开。
+ *
+ * 回归：URLDecoder 会把 '+' 解成空格，导致 {@code plus+name.zip} 被解析为
+ * {@code plus name.zip}（两种写法都找不到文件）。
+ */
+ @Test
+ void adminDownloadServesFileWithLiteralPlusInName(@TempDir Path root) throws Exception {
+ Path galleryRoot = Files.createDirectories(root.resolve("gallery"));
+ byte[] content = "plus-payload".getBytes(StandardCharsets.UTF_8);
+ Path archive = Files.write(galleryRoot.resolve("plus+name.zip"), content);
+ String previousRoot = lion.Config.Config.adminDownloadRoot;
+ Fixture fixture = Fixture.start(new GalleryDownloadHandler(
+ new GalleryArchiveIndex(galleryRoot.toFile(), 1_000L)), "plus-code", galleryRoot.toString());
+ try {
+ // 管理员通道按真实路径直取，因此用绝对路径请求。
+ // 关键点：路径里保持「字面量 +」（不编码成 %2B）——若用 URLDecoder 解码路径，
+ // '+' 会被解成空格，从而 404。
+ String requestPath = archive.toRealPath().toString();
+ Response response = fixture.get(requestPath + "?AuthCode=plus-code");
+ assertEquals(200, response.status(), "字面量 '+' 文件应可下载");
+ assertArrayEquals(content, response.body());
+ } finally {
+ fixture.close();
+ lion.Config.Config.adminDownloadCode = "";
+ lion.Config.Config.adminDownloadRoot = previousRoot;
+ }
+ }
+
+ /** 完整下载必须回 200 且 Content-Length 等于文件长度（旧实现恒为 206）。 */
+ @Test
+ void fullDownloadReturnsOkWithExactLength(@TempDir Path root) throws Exception {
+ byte[] payload = "0123456789".getBytes(StandardCharsets.UTF_8);
+ Fixture fixture = Fixture.start(root, "Title [123]", payload);
+ try {
+ Response response = fixture.get("/file?AuthCode=x&gid=123");
+ assertEquals(200, response.status());
+ assertEquals("10", response.header("content-length"));
+ assertArrayEquals(payload, response.body());
+ assertNull(response.header("content-range"), "完整下载不应带 Content-Range");
+ assertEquals("bytes", response.header("accept-ranges"));
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 普通区间 Range：bytes=1-3。 */
+ @Test
+ void plainRangeReturnsPartialContent(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "0123456789".getBytes(StandardCharsets.UTF_8));
+ try {
+ Response response = fixture.get("/file?AuthCode=x&gid=123", "bytes=1-3");
+ assertEquals(206, response.status());
+ assertEquals("bytes 1-3/10", response.header("content-range"));
+ assertEquals("3", response.header("content-length"));
+ assertEquals("123", response.bodyAsString());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 后缀区间 bytes=-4 取最后 4 字节（旧实现不支持）。 */
+ @Test
+ void suffixRangeReturnsLastBytes(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "0123456789".getBytes(StandardCharsets.UTF_8));
+ try {
+ Response response = fixture.get("/file?AuthCode=x&gid=123", "bytes=-4");
+ assertEquals(206, response.status());
+ assertEquals("bytes 6-9/10", response.header("content-range"));
+ assertEquals("6789", response.bodyAsString());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 开放结尾 bytes=2- 取到文件末尾。 */
+ @Test
+ void openEndedRangeGoesToFileEnd(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "0123456789".getBytes(StandardCharsets.UTF_8));
+ try {
+ Response response = fixture.get("/file?AuthCode=x&gid=123", "bytes=2-");
+ assertEquals(206, response.status());
+ assertEquals("bytes 2-9/10", response.header("content-range"));
+ assertEquals("23456789", response.bodyAsString());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 超出末端的区间必须夹到文件末尾。 */
+ @Test
+ void oversizedRangeIsClampedToFileEnd(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "0123456789".getBytes(StandardCharsets.UTF_8));
+ try {
+ Response response = fixture.get("/file?AuthCode=x&gid=123", "bytes=4-999999");
+ assertEquals(206, response.status());
+ assertEquals("bytes 4-9/10", response.header("content-range"));
+ assertEquals("6", response.header("content-length"));
+ assertEquals("456789", response.bodyAsString());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 起点越界回 416。 */
+ @Test
+ void rangeStartingBeyondFileIsNotSatisfiable(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "0123456789".getBytes(StandardCharsets.UTF_8));
+ try {
+ assertEquals(416, fixture.get("/file?AuthCode=x&gid=123", "bytes=999-1000").status());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 语法无法解析的 Range 按 RFC 忽略，退回完整 200。 */
+ @Test
+ void malformedRangeFallsBackToFullEntity(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "0123456789".getBytes(StandardCharsets.UTF_8));
+ try {
+ Response response = fixture.get("/file?AuthCode=x&gid=123", "bytes=abc-def");
+ assertEquals(200, response.status());
+ assertEquals("10", response.header("content-length"));
+ assertEquals("0123456789", response.bodyAsString());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** HEAD 必须给出 Content-Length，但不回正文。 */
+ @Test
+ void headReturnsHeadersWithoutBody(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "0123456789".getBytes(StandardCharsets.UTF_8));
+ try {
+ Response response = fixture.request("HEAD", "/file?AuthCode=x&gid=123");
+ assertEquals(200, response.status());
+ assertEquals("10", response.header("content-length"));
+ assertEquals(0, response.body().length, "HEAD 不得返回正文");
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** HEAD 带 Range 时同样给出区间长度且不回正文。 */
+ @Test
+ void headWithRangeReportsPartialLength(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "0123456789".getBytes(StandardCharsets.UTF_8));
+ try {
+ Response response = fixture.request("HEAD", "/file?AuthCode=x&gid=123", "bytes=2-4");
+ assertEquals(206, response.status());
+ assertEquals("3", response.header("content-length"));
+ assertEquals("bytes 2-4/10", response.header("content-range"));
+ assertEquals(0, response.body().length);
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 归档响应带 Content-Disposition，且文件名中的引号已被清理。 */
+ @Test
+ void archiveResponseCarriesSanitizedDisposition(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title \"quoted\" [123]", "abc".getBytes(StandardCharsets.UTF_8));
+ try {
+ Response response = fixture.get("/file?AuthCode=x&gid=123");
+ assertEquals(200, response.status());
+ String disposition = response.header("content-disposition");
+ assertNotNull(disposition);
+ assertTrue(disposition.startsWith("attachment; filename=\""));
+ assertFalse(disposition.contains("\"quoted\""), "文件名里的引号必须被清理：" + disposition);
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 不存在的 gid 回 404。 */
+ @Test
+ void unknownGidIsNotFound(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "abc".getBytes(StandardCharsets.UTF_8));
+ try {
+ assertEquals(404, fixture.get("/file?AuthCode=x&gid=999").status());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 缺少 AuthCode 参数一律 403。 */
+ @Test
+ void missingAuthCodeIsForbidden(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "abc".getBytes(StandardCharsets.UTF_8));
+ try {
+ assertEquals(403, fixture.get("/file?gid=123").status());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 非 GET/HEAD 方法回 501（保持旧行为）。 */
+ @Test
+ void unsupportedMethodIsNotImplemented(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "abc".getBytes(StandardCharsets.UTF_8));
+ try {
+ assertEquals(501, fixture.request("POST", "/file?AuthCode=x&gid=123").status());
+ } finally {
+ fixture.close();
+ }
+ }
+
+ /** 关闭服务后端口必须立即释放，且可重复关闭。 */
+ @Test
+ void closeReleasesPortAndIsIdempotent(@TempDir Path root) throws Exception {
+ Fixture fixture = Fixture.start(root, "Title [123]", "abc".getBytes(StandardCharsets.UTF_8));
+ int port = fixture.serverPort;
+
+ fixture.close();
+ fixture.close(); // 重复关闭不得抛异常
+
+ try (ServerSocket rebind = new ServerSocket(port)) {
+ assertEquals(port, rebind.getLocalPort(), "关闭后端口必须可重新绑定");
+ }
+ }
+
// ---- 测试脚手架 ----
- /** 在守护线程里接受一个连接并交给真实的请求处理器。 */
- private static Thread startHandler(ServerSocket server) {
- return startHandler(server, new AtomicReference<>());
- }
+ /** 起一个真实端口的下载服务，归档放在临时画廊根目录下。 */
+ private static final class Fixture {
+ private final HttpFileServer server;
+ private final int serverPort;
- private static Thread startHandler(ServerSocket server, AtomicReference 全部离线运行：不绑定端口，也不外呼真实主站。
+ */
+class ReliableTaskServiceTest {
+
+ /** 相同 operationId 重复执行不得产生第二次副作用，且回放同一结果码。 */
+ @Test
+ void repeatedDownloadCommandReusesReceipt(@TempDir Path root) throws Exception {
+ Harness harness = new Harness(root);
+ TaskCommandMessage command = download("op-dl-1", 80001, "sample [80001]");
+
+ assertEquals(NodeErrorCode.SUCCESS, harness.service.handleCommand(command));
+ GalleryTask first = harness.queue.get(80001);
+ assertNotNull(first);
+
+ // 更换 name 再发同一 operationId：必须回放回执，不得改写已受理的任务。
+ TaskCommandMessage retry = download("op-dl-1", 80001, "changed-name [80001]");
+ assertEquals(NodeErrorCode.SUCCESS, harness.service.handleCommand(retry));
+ assertSame(first, harness.queue.get(80001), "重复命令不得替换已受理的任务对象");
+ assertEquals("sample [80001]", harness.queue.get(80001).getName());
+ }
+
+ /** 删除成功是终态：重复删除直接回放，绝不再次执行删除副作用。 */
+ @Test
+ void repeatedDeleteCommandDoesNotRerunSideEffect(@TempDir Path root) throws Exception {
+ Harness harness = new Harness(root);
+ Path gallery = Files.createDirectories(harness.storagePath.resolve("sample [80002]"));
+
+ TaskCommandMessage delete = delete("op-del-1", 80002, "sample [80002]");
+ assertEquals(NodeErrorCode.SUCCESS, harness.service.handleCommand(delete));
+ assertFalse(Files.exists(gallery));
+
+ // 重新创建同名目录，再发同一 operationId：若回执失效会把它删掉，这里必须保留。
+ Files.createDirectories(gallery);
+ assertEquals(NodeErrorCode.SUCCESS, harness.service.handleCommand(delete));
+ assertTrue(Files.exists(gallery), "重复删除命令不得再次执行副作用");
+ }
+
+ /** 压缩中的目标必须回可重试的「忙」，且不得写入永久回执。 */
+ @Test
+ void deleteWhileCompressingIsRetryable(@TempDir Path root) throws Exception {
+ Harness harness = new Harness(root);
+ Path gallery = Files.createDirectories(harness.storagePath.resolve("sample [80003]"));
+ harness.downloadCheck.activeCompressions.add(80003);
+
+ byte result = harness.service.handleCommand(delete("op-del-busy", 80003, "sample [80003]"));
+
+ assertEquals(NodeErrorCode.BUSY_COMPRESSING, result);
+ assertNull(harness.service.journal().receipt("op-del-busy"), "可重试失败不得记为永久回执");
+ assertTrue(Files.exists(gallery), "忙时不得动磁盘");
+
+ // 压缩结束后同一 operationId 重试应当成功。
+ harness.downloadCheck.activeCompressions.remove(80003);
+ assertEquals(NodeErrorCode.SUCCESS, harness.service.handleCommand(delete("op-del-busy", 80003, "sample [80003]")));
+ assertFalse(Files.exists(gallery));
+ }
+
+ /**
+ * ACK 只清理压缩完成的终态任务。
+ *
+ * 这是根因级缺陷：若按「报告里出现的 status」无条件删除，正在下载的任务会在主站
+ * 刚收到进度报告后就被节点遗忘，下载随后无人推进，任务永久停在未完成。
+ */
+ @Test
+ void ackKeepsInProgressTaskAndClearsOnlyCompleted(@TempDir Path root) throws Exception {
+ Harness harness = new Harness(root);
+ harness.service.handleCommand(download("op-a", 80010, "in-progress [80010]"));
+ harness.service.handleCommand(download("op-b", 80011, "finished [80011]"));
+ harness.queue.get(80011).setStatus(GalleryTask.COMPRESS_COMPLETE);
+
+ TaskJournal.ReportSnapshot snapshot = harness.service.buildReport().orElseThrow();
+ assertEquals(2, snapshot.tasks.size());
+
+ harness.service.applyAck(snapshot.reportId, NodeErrorCode.SUCCESS);
+
+ assertTrue(harness.queue.containsKey(80010), "下载中的任务不得因 ACK 被移除");
+ assertTrue(harness.service.journal().tasks().containsKey(80010), "日志里也应保留进行中的任务");
+ assertFalse(harness.queue.containsKey(80011), "压缩完成的任务应被清理");
+ assertFalse(harness.service.journal().tasks().containsKey(80011));
+ }
+
+ /** 失败或未知的 ACK 不得移除任何任务，也不得丢掉待发快照。 */
+ @Test
+ void rejectedAndDuplicateAcksKeepEverything(@TempDir Path root) throws Exception {
+ Harness harness = new Harness(root);
+ harness.service.handleCommand(download("op-c", 80020, "done [80020]"));
+ harness.queue.get(80020).setStatus(GalleryTask.COMPRESS_COMPLETE);
+ TaskJournal.ReportSnapshot snapshot = harness.service.buildReport().orElseThrow();
+
+ assertTrue(harness.service.applyAck(snapshot.reportId, NodeErrorCode.IO_ERROR).isEmpty(),
+ "未提交的 ACK 不得清理任务");
+ assertTrue(harness.service.pendingReports().stream().anyMatch(r -> r.reportId.equals(snapshot.reportId)),
+ "失败 ACK 必须保留快照等待重试");
+ assertTrue(harness.queue.containsKey(80020));
+
+ // 重复 ACK：第一次清理，第二次无副作用。
+ assertEquals(1, harness.service.applyAck(snapshot.reportId, NodeErrorCode.SUCCESS).size());
+ assertTrue(harness.service.applyAck(snapshot.reportId, NodeErrorCode.SUCCESS).isEmpty(),
+ "重复 ACK 不得重复清理或报错");
+ }
+
+ /** 迟到的 ACK 碰上已经又前进的任务时，必须保留该任务。 */
+ @Test
+ void lateAckDoesNotRemoveTaskThatAdvanced(@TempDir Path root) throws Exception {
+ Harness harness = new Harness(root);
+ harness.service.handleCommand(download("op-late", 80030, "advancing [80030]"));
+ TaskJournal.ReportSnapshot snapshot = harness.service.buildReport().orElseThrow();
+
+ // 报告发出后任务又前进到完成（尚未重新落盘），此时迟到的 ACK 只代表更早的状态。
+ harness.queue.get(80030).setStatus(GalleryTask.COMPRESS_COMPLETE);
+ harness.queue.get(80030).setProceeding(99);
+
+ harness.service.applyAck(snapshot.reportId, NodeErrorCode.SUCCESS);
+
+ assertTrue(harness.queue.containsKey(80030), "ACK 内容与运行期状态不一致时不得删除任务");
+ }
+
+ /** 重启后必须恢复任务与待 ACK 报告，并把中断的压缩退回可重试状态。 */
+ @Test
+ void restartRestoresTasksReportsAndResetsInterruptedCompression(@TempDir Path root) throws Exception {
+ Harness first = new Harness(root);
+ first.service.handleCommand(download("op-r", 80040, "resume [80040]"));
+ TaskJournal.ReportSnapshot snapshot = first.service.buildReport().orElseThrow();
+ // 中断在压缩中途：重启后应回到可重试的下载完成态。
+ first.queue.get(80040).setStatus(GalleryTask.COMPRESSING);
+ first.service.refreshJournalFromQueue();
+
+ Harness restarted = new Harness(root, true);
+
+ assertTrue(restarted.queue.containsKey(80040), "重启应恢复任务");
+ assertEquals(GalleryTask.DOWNLOAD_COMPLETE, restarted.queue.get(80040).getStatus(),
+ "中断的压缩应退回可重试状态");
+ assertTrue(restarted.service.pendingReports().stream()
+ .anyMatch(r -> r.reportId.equals(snapshot.reportId)),
+ "重启应恢复待 ACK 报告以便重发");
+ assertNotNull(restarted.service.journal().receipt("op-r"), "命令回执必须跨重启保留");
+ }
+
+ /** 命令缺 operationId 或 action 非法时回不可重试的无效命令，且不产生回执。 */
+ @Test
+ void invalidCommandsAreRejectedWithoutReceipt(@TempDir Path root) throws Exception {
+ Harness harness = new Harness(root);
+
+ TaskCommandMessage noId = download(null, 80050, "x [80050]");
+ assertEquals(NodeErrorCode.INVALID_COMMAND, harness.service.handleCommand(noId));
+
+ TaskCommandMessage unknownAction = download("op-u", 80051, "x [80051]");
+ unknownAction.setAction("FROBNICATE");
+ assertEquals(NodeErrorCode.INVALID_COMMAND, harness.service.handleCommand(unknownAction));
+
+ assertTrue(harness.service.journal().receipts().isEmpty(), "无效命令不得留下回执");
+ }
+
+ /** 一个最小离线装配：显式目录、无调度器、无网络。 */
+ static class Harness {
+ final Map 这些用例锁定的是一条不变式：内存状态永不领先磁盘。任何一次落盘失败之后，内存里能看到的
+ * 内容必须与磁盘上的最后一份一致，否则重启会丢命令回执，主站重试将重复执行副作用。
+ */
+class TaskJournalTest {
+
+ @Test
+ void failedCommitLeavesMemoryAtLastDurableState(@TempDir Path root) throws Exception {
+ TaskJournal journal = new TaskJournal(root);
+ GalleryTask task = task(70001, GalleryTask.DOWNLOADING, 3);
+ journal.bumpAndPutTask(task, "op-1");
+ long durableRevision = journal.revision();
+
+ // 用一个一定失败的目录替换：把日志目录指向一个普通文件，任何写入都会抛 IOException。
+ Path blocked = root.resolve("blocked");
+ Files.writeString(blocked, "not a directory");
+ TaskJournal broken = new TaskJournal(blocked);
+ assertThrows(IOException.class,
+ () -> broken.bumpAndPutTask(task(70002, GalleryTask.DOWNLOADING, 1), "op-2"));
+ assertTrue(broken.tasks().isEmpty(), "写盘失败后内存不得留有未落盘的任务");
+ assertEquals(0, broken.revision(), "失败不得推进 revision");
+
+ // 原日志不受影响，仍然是那份已成功落盘的内容。
+ assertEquals(durableRevision, journal.revision());
+ assertEquals(1, journal.tasks().size());
+ }
+
+ /** 成功的写盘必须真的出现在磁盘上，重新加载后内容一致。 */
+ @Test
+ void reloadReturnsExactlyWhatWasCommitted(@TempDir Path root) throws Exception {
+ TaskJournal journal = new TaskJournal(root);
+ journal.bumpAndPutTask(task(70010, GalleryTask.COMPRESS_COMPLETE, 12), "op-a");
+ journal.putReceipt(TaskJournal.CommandReceipt.of("op-a", "DOWNLOAD", 70010, "name", NodeErrorCode.SUCCESS));
+
+ TaskJournal reloaded = new TaskJournal(root);
+ reloaded.load();
+
+ assertEquals(1, reloaded.tasks().size());
+ assertEquals(70010, reloaded.tasks().values().iterator().next().gid);
+ assertEquals(NodeErrorCode.SUCCESS, reloaded.receipt("op-a").result);
+ assertEquals(journal.revision(), reloaded.revision());
+ }
+
+ /** 高于本程序支持的 schemaVersion 必须拒绝启动，而不是按旧规则硬读。 */
+ @Test
+ void refusesNewerSchemaVersion(@TempDir Path root) throws Exception {
+ Path directory = Files.createDirectories(root.resolve("journal"));
+ Files.writeString(directory.resolve(TaskJournal.STATE_FILE),
+ "{\"schemaVersion\": 99, \"revision\": 5, \"tasks\": {}, \"commandReceipts\": {}, \"pendingReports\": {}}");
+
+ IOException failure = assertThrows(IOException.class, () -> new TaskJournal(directory).load());
+ assertTrue(failure.getMessage().contains("schemaVersion"), "错误信息应指出是版本问题");
+ }
+
+ /** 损坏的日志文件不得被当作空日志吞掉。 */
+ @Test
+ void refusesCorruptJournal(@TempDir Path root) throws Exception {
+ Path directory = Files.createDirectories(root.resolve("journal"));
+ Files.writeString(directory.resolve(TaskJournal.STATE_FILE), "{not json");
+
+ assertThrows(IOException.class, () -> new TaskJournal(directory).load());
+ }
+
+ /** 快照必须拷贝任务字段：之后任务继续变化不得改写已生成的报告内容。 */
+ @Test
+ void reportSnapshotCopiesTaskFields(@TempDir Path root) throws Exception {
+ TaskJournal journal = new TaskJournal(root);
+ TaskJournal.JournalTask stored = journal.bumpAndPutTask(task(70020, GalleryTask.DOWNLOADING, 1), "op-x");
+
+ TaskJournal.ReportSnapshot snapshot = TaskJournal.ReportSnapshot.of(
+ "report-1", stored.revision, List.of(stored), Map.of(70020, "op-x"));
+
+ // 模拟扫描继续推进同一个任务对象。
+ GalleryTask advanced = stored.toGalleryTask();
+ advanced.setStatus(GalleryTask.COMPRESS_COMPLETE);
+ advanced.setProceeding(44);
+
+ TaskJournal.JournalTask inSnapshot = snapshot.tasks.get(0);
+ assertEquals(GalleryTask.DOWNLOADING, inSnapshot.status, "快照内容不得被后续变化改写");
+ assertEquals(1, inSnapshot.proceeding);
+
+ // 而且对外转换出的协议对象也是独立副本。
+ GalleryTask[] wire = snapshot.toGalleryTasks();
+ wire[0].setStatus(GalleryTask.COMPRESS_COMPLETE);
+ assertEquals(GalleryTask.DOWNLOADING, snapshot.tasks.get(0).status);
+ }
+
+ private static GalleryTask task(int gid, byte status, int proceeding) {
+ GalleryTask task = new GalleryTask();
+ task.setGid(gid);
+ task.setName("sample [" + gid + "]");
+ task.setStatus(status);
+ task.setProceeding(proceeding);
+ return task;
+ }
+}
diff --git a/src/test/java/lion/StorageNodeMessageHandlingTest.java b/src/test/java/lion/StorageNodeMessageHandlingTest.java
index 3f0531f..79febb7 100644
--- a/src/test/java/lion/StorageNodeMessageHandlingTest.java
+++ b/src/test/java/lion/StorageNodeMessageHandlingTest.java
@@ -9,8 +9,13 @@ import lion.Message.Main.DeleteGalleryMessage;
import lion.Message.Main.DownloadPostMessage;
import lion.Message.Main.DownloadStatusMessage;
import lion.Message.Main.IdentityMessage;
+import lion.Message.Main.ProtocolHelloMessage;
import lion.Message.Main.ResponseMessage;
import lion.Message.Main.SubscriptionSnapshotMessage;
+import lion.Message.Main.TaskCommandMessage;
+import lion.Message.Main.TaskReportAckMessage;
+import lion.Message.Main.TaskReportMessage;
+import lion.Service.NodeErrorCode;
import lion.Service.SubscriptionSnapshotStore;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
@@ -46,21 +51,278 @@ class StorageNodeMessageHandlingTest {
storageNode.storagePath = storageRoot + "/";
SubscriptionSnapshotStore store =
new SubscriptionSnapshotStore(subscriptionRoot, "test-secret", 3600, 1024 * 1024);
- node = new storageNode(store);
+ // 离线构造：不绑定端口、不唤醒真实主站、不启动调度线程。
+ node = new storageNode(store, tempDir.resolve("task-journal"), false);
// 订阅同步默认关闭：快照消息必须被拒绝，也不触发异步落盘。
Config.subscriptionSyncEnabled = false;
}
@AfterAll
static void stopNode() {
- node.subscriptionApplyExecutor.shutdownNow();
- node.rekickExecutor.shutdownNow();
+ // 统一释放：压缩调度、通知、重连与可靠执行器都不得在测试结束后残留线程。
+ node.shutdown();
}
@BeforeEach
- void resetState() {
+ void resetState() throws Exception {
node.queue.clear();
node.tempQueue.clear();
+ node.reportSends.clear();
+ // 每个用例用独立的任务日志：共享一份日志会让上一个用例的任务与回执泄漏到下一个。
+ node.reliableTaskService = new lion.Service.ReliableTaskService(
+ Files.createDirectories(tempDir.resolve("journal-" + java.util.UUID.randomUUID())),
+ node.queue, node.downloadCheckService, Path.of(storageNode.storagePath));
+ node.reportRetryIntervalMillis = 5_000L;
+ }
+
+ /**
+ * v2 能力协商：收到声明 version2 的 hello 后，节点回一条同样的 hello，并切换到可靠上报。
+ * 旧主站因此只会在自己先发起协商时才看到 type10。
+ */
+ @Test
+ void protocolHelloNegotiatesAndRepliesHello() {
+ EmbeddedChannel channel = nodeChannel();
+ try {
+ channel.writeInbound(ProtocolHelloMessage.v2());
+
+ ProtocolHelloMessage reply = channel.readOutbound();
+ assertNotNull(reply, "确认对端支持 v2 后必须回 hello");
+ assertEquals(ProtocolHelloMessage.VERSION, reply.getProtocolVersion());
+ assertTrue(reply.getFeatures().contains(ProtocolHelloMessage.RELIABLE_TASK_V2));
+ assertTrue(node.reliableTaskService.reporter().reliableV2(), "应切换到可靠任务协议");
+ } finally {
+ close(channel);
+ }
+ }
+
+ /**
+ * 未声明 v2 的 hello（例如更老或未知版本）不得让节点升级，也不得回包或断开连接。
+ * 这条用例覆盖「旧对端发来未知 hello 时不能退出」的兼容要求。
+ */
+ @Test
+ void legacyHelloDoesNotUpgradeOrBreakConnection() {
+ EmbeddedChannel channel = nodeChannel();
+ try {
+ ProtocolHelloMessage legacy = new ProtocolHelloMessage();
+ legacy.setProtocolVersion(1);
+ legacy.setFeatures(java.util.List.of("something-else"));
+
+ channel.writeInbound(legacy);
+
+ assertNull(channel.readOutbound(), "未达成 v2 时不应回 hello");
+ assertFalse(node.reliableTaskService.reporter().reliableV2());
+ assertTrue(channel.isActive(), "未知 hello 不得导致连接异常");
+ } finally {
+ close(channel);
+ }
+ }
+
+ /** 任务命令：先回执，再由可靠服务持久化任务与回执。 */
+ @Test
+ void taskCommandIsAppliedAndReceipted() throws Exception {
+ EmbeddedChannel channel = nodeChannel();
+ try {
+ channel.writeInbound(ProtocolHelloMessage.v2());
+ drainOutbound(channel);
+
+ TaskCommandMessage command = new TaskCommandMessage();
+ command.messageId = 77;
+ command.setOperationId("op-node-1");
+ command.setAction(TaskCommandMessage.ACTION_DOWNLOAD);
+ command.setGid(91001);
+ command.setName("sample [91001]");
+
+ channel.writeInbound(command);
+ awaitReliableExecutor();
+
+ ResponseMessage response = readResponse(channel, 77);
+ assertNotNull(response, "命令必须按原 messageId 回执，否则主站无法配对");
+ assertEquals(NodeErrorCode.SUCCESS, response.getResult());
+ assertNotNull(node.reliableTaskService.journal().receipt("op-node-1"),
+ "成功命令必须落盘回执以便重试幂等");
+ } finally {
+ close(channel);
+ }
+ }
+
+ /** 报告 ACK：压缩完成的任务在收到成功 ACK 后才从队列移除。 */
+ @Test
+ void reportAckClearsCompletedTaskOnlyAfterSuccess() throws Exception {
+ EmbeddedChannel channel = nodeChannel();
+ try {
+ channel.writeInbound(identity("lionwebsite"));
+ channel.writeInbound(ProtocolHelloMessage.v2());
+ drainOutbound(channel);
+
+ GalleryTask done = task(91002);
+ done.setStatus(GalleryTask.COMPRESS_COMPLETE);
+ node.queue.put(91002, done);
+ // 模拟扫描已把最新进度落盘（生产里由 flushReport 先做这一步）。
+ node.reliableTaskService.refreshJournalFromQueue();
+
+ node.flushReport();
+ TaskReportMessage report = readReport(channel);
+ assertNotNull(report, "协商 v2 后应发送可靠报告");
+ assertFalse(node.queue.isEmpty(), "发出报告本身不得移除任务");
+
+ TaskReportAckMessage ack = new TaskReportAckMessage();
+ ack.setReportId(report.getReportId());
+ ack.setResult(TaskReportAckMessage.RESULT_SUCCESS);
+ channel.writeInbound(ack);
+ awaitReliableExecutor();
+
+ assertTrue(node.queue.isEmpty(), "成功后应清理压缩完成的任务");
+ assertFalse(node.reliableTaskService.journal().tasks().containsKey(91002));
+ } finally {
+ close(channel);
+ }
+ }
+
+ /**
+ * 同一连接上 ACK 丢失时必须能重发：TCP 未断、通道引用未变，也要在重发间隔后再次发送，
+ * 且 reportId 与 revision 保持稳定。
+ */
+ @Test
+ void pendingReportIsResentOnSameChannelUntilAcked() throws Exception {
+ EmbeddedChannel channel = nodeChannel();
+ try {
+ channel.writeInbound(identity("lionwebsite"));
+ channel.writeInbound(ProtocolHelloMessage.v2());
+ drainOutbound(channel);
+
+ GalleryTask done = task(91003);
+ done.setStatus(GalleryTask.COMPRESS_COMPLETE);
+ node.queue.put(91003, done);
+ node.reliableTaskService.refreshJournalFromQueue();
+
+ node.flushReport();
+ TaskReportMessage first = readReport(channel);
+ assertNotNull(first);
+
+ // 同一通道、ACK 未到：间隔内不重发。
+ node.flushReport();
+ assertNull(readReport(channel), "重发间隔内不应重复发送");
+
+ // 超过间隔后必须重发，且身份不变。
+ node.reportRetryIntervalMillis = 0;
+ node.flushReport();
+ TaskReportMessage resent = readReport(channel);
+ assertNotNull(resent, "超出重发间隔后必须重发，否则 ACK 丢失会永久卡住");
+ assertEquals(first.getReportId(), resent.getReportId(), "重发必须复用同一 reportId");
+ assertEquals(first.getRevision(), resent.getRevision(), "重发必须复用同一 revision");
+
+ // ACK 到达后停止重发。
+ TaskReportAckMessage ack = new TaskReportAckMessage();
+ ack.setReportId(first.getReportId());
+ ack.setResult(TaskReportAckMessage.RESULT_SUCCESS);
+ channel.writeInbound(ack);
+ awaitReliableExecutor();
+ node.flushReport();
+ assertNull(readReport(channel), "确认后不得再重发");
+ } finally {
+ close(channel);
+ }
+ }
+
+ /** 写入通道失败时必须清除发送登记，使下一轮立即重试。 */
+ @Test
+ void failedChannelWriteAllowsImmediateRetry() throws Exception {
+ FailingChannel channel = new FailingChannel(node.new MyChannelInboundHandlerAdapter(node.tempQueue));
+ try {
+ channel.writeInbound(identity("lionwebsite"));
+ channel.writeInbound(ProtocolHelloMessage.v2());
+ drainOutbound(channel);
+ channel.failWrites = true;
+
+ GalleryTask done = task(91004);
+ done.setStatus(GalleryTask.COMPRESS_COMPLETE);
+ node.queue.put(91004, done);
+ node.reliableTaskService.refreshJournalFromQueue();
+ node.reportRetryIntervalMillis = 60_000;
+
+ node.flushReport();
+ channel.runPendingTasks();
+
+ assertTrue(node.reportSends.isEmpty(), "写入失败后不得留下发送登记");
+ assertTrue(node.queue.containsKey(91004), "写入失败绝不能让任务消失");
+ } finally {
+ channel.finishAndReleaseAll();
+ }
+ }
+
+ /** 通道断开后必须复位协商结果，避免向未协商的对端发送 v2 消息。 */
+ @Test
+ void disconnectResetsNegotiation() {
+ EmbeddedChannel channel = nodeChannel();
+ channel.writeInbound(ProtocolHelloMessage.v2());
+ assertTrue(node.reliableTaskService.reporter().reliableV2());
+
+ close(channel);
+
+ assertFalse(node.reliableTaskService.reporter().reliableV2(), "断线后应重新协商");
+ }
+
+ /**
+ * 队列为空、扫描也无变化时，仍必须重试待确认报告。
+ *
+ * 历史缺陷：待发报告的路径都挂在「有任务变化」的分支后面，一旦任务不再变化
+ * （例如下载已结束、或应用刚重启且队列为空），ACK 丢失的报告就再也不会被重发。
+ */
+ @Test
+ void pendingReportIsRetriedEvenWhenQueueIsEmpty() throws Exception {
+ EmbeddedChannel channel = nodeChannel();
+ try {
+ channel.writeInbound(identity("lionwebsite"));
+ channel.writeInbound(ProtocolHelloMessage.v2());
+ drainOutbound(channel);
+
+ GalleryTask done = task(91005);
+ done.setStatus(GalleryTask.COMPRESS_COMPLETE);
+ node.queue.put(91005, done);
+ node.reliableTaskService.refreshJournalFromQueue();
+ node.reportRetryIntervalMillis = 0;
+
+ node.flushReport();
+ assertNotNull(readReport(channel));
+
+ // 模拟主站已把任务标为完成并从节点队列移除，但 ACK 仍未到达。
+ node.queue.clear();
+ node.mainThread();
+
+ TaskReportMessage resent = readReport(channel);
+ assertNotNull(resent, "队列为空也必须重试待确认报告");
+ } finally {
+ close(channel);
+ }
+ }
+
+ /**
+ * 旧协议路径下，写入失败时不得清空已完成任务。
+ *
+ * 旧协议没有 ACK，唯一的安全边界就是「写成功」。若在写失败时照旧把压缩完成的任务
+ * 出队，这份完成状态既没送达主站也不会再重发，任务会永久停在未完成。
+ */
+ @Test
+ void legacyReportKeepsTaskWhenWriteFails() {
+ FailingChannel channel = new FailingChannel(node.new MyChannelInboundHandlerAdapter(node.tempQueue));
+ try {
+ channel.writeInbound(identity("lionwebsite"));
+ drainOutbound(channel);
+ channel.failWrites = true;
+
+ GalleryTask done = task(91006);
+ done.setStatus(GalleryTask.COMPRESS_COMPLETE);
+ node.queue.put(91006, done);
+
+ // 未协商 v2：走旧上报路径。
+ assertFalse(node.reliableTaskService.reporter().reliableV2());
+ node.mainThread();
+ channel.runPendingTasks();
+
+ assertTrue(node.queue.containsKey(91006), "写失败时任务必须保留在队列中等待重发");
+ } finally {
+ channel.finishAndReleaseAll();
+ }
}
/**
@@ -261,6 +523,31 @@ class StorageNodeMessageHandlingTest {
return new EmbeddedChannel(node.new MyChannelInboundHandlerAdapter(node.tempQueue));
}
+ /** 等待可靠执行器把当前排队的任务处理完，使异步回包/落盘对测试可见。 */
+ private static void awaitReliableExecutor() throws Exception {
+ node.reliableExecutor.submit(() -> { }).get(5, java.util.concurrent.TimeUnit.SECONDS);
+ // 执行器是单线程 FIFO：再提交一个空任务并等它返回，即代表之前的任务都已完成。
+ node.reliableExecutor.submit(() -> { }).get(5, java.util.concurrent.TimeUnit.SECONDS);
+ }
+
+ /** 读到指定 messageId 的响应消息，忽略其它出站内容。 */
+ private static ResponseMessage readResponse(EmbeddedChannel channel, int messageId) {
+ Object outbound;
+ while ((outbound = channel.readOutbound()) != null)
+ if (outbound instanceof ResponseMessage response && response.messageId == messageId)
+ return response;
+ return null;
+ }
+
+ /** 读出第一条可靠报告；容忍排在其前面的其它出站消息。 */
+ private static TaskReportMessage readReport(EmbeddedChannel channel) {
+ Object outbound;
+ while ((outbound = channel.readOutbound()) != null)
+ if (outbound instanceof TaskReportMessage report)
+ return report;
+ return null;
+ }
+
private static GalleryTask task(int gid) {
GalleryTask task = new GalleryTask();
task.setGid(gid);
@@ -294,4 +581,25 @@ class StorageNodeMessageHandlingTest {
private static void close(EmbeddedChannel channel) {
channel.finishAndReleaseAll();
}
+
+ /**
+ * 可注入写失败的 EmbeddedChannel。
+ *
+ * 用于验证「TCP 未断但写入失败」这一分支：写入失败必须清除发送登记，
+ * 让下一轮扫描立刻重发，而不是把报告当成已送达而永久留在待确认状态。
+ */
+ static class FailingChannel extends EmbeddedChannel {
+ volatile boolean failWrites;
+
+ FailingChannel(io.netty.channel.ChannelHandler... handlers) {
+ super(handlers);
+ }
+
+ @Override
+ protected void handleOutboundMessage(Object msg) {
+ if (failWrites)
+ throw new io.netty.channel.ChannelException("injected write failure");
+ super.handleOutboundMessage(msg);
+ }
+ }
}
+ *
+ *
+ *
+ *
+ *
+ * 501 Not Implemented
");
+
+ String authCode = request.parameter("AuthCode");
+ if (authCode == null)
+ return HttpResponseModel.text(403, "text/html", "403 Forbidden
");
+
+ File file = isAdminRequest(authCode) ? resolveAdminRequest(request) : resolveGalleryRequest(request);
+ if (file == null || !file.exists() || !file.isFile() || !file.canRead())
+ return HttpResponseModel.text(404, "text/html", "404 Not Found
");
+
+ String fileName = file.getName();
+ log.info("提供下载:{}", file.getAbsolutePath());
+ return FileResponseFactory.forFile(request.rangeHeader(), file, fileName);
+ }
+
+ private File resolveAdminRequest(HttpRequestModel request) {
+ File file = resolveAdminFile(request.decodedPath());
+ if (file == null)
+ log.warn("管理员下载越界，已拒绝:{}", request.decodedPath());
+ return file;
+ }
+
+ private File resolveGalleryRequest(HttpRequestModel request) {
+ String gid = request.parameter("gid");
+ File file = gid == null ? null : archiveIndex.find(gid);
+ if (file != null)
+ return file;
+
+ // 兼容没有 gid 参数的旧下载链接：按链接中的文件名找回归档。
+ String requestPath = request.decodedPath();
+ if (requestPath != null && requestPath.contains(".")) {
+ String name = requestPath.substring(0, requestPath.lastIndexOf('.'));
+ return new File(LEGACY_GALLERY_ROOT + name + "/" + name + ".zip");
+ }
+ return FILE_NOT_FOUND;
+ }
+
+ /**
+ * 该请求是否走管理员直取文件通道。
+ *
+ * 500 Internal Server Error
");
+ }
+ if (response.closeWithoutResponse()) {
+ // 不调用 sendResponseHeaders，try-with-resources 关掉交换即直接断开 TCP。
+ log.info("拒绝来源 {} 的请求: {}", exchange.getRemoteAddress(), exchange.getRequestURI());
+ return;
+ }
+ write(exchange, response);
+ } catch (IOException e) {
+ // 客户端提前断开属于正常情况，记录后不再尝试写响应。
+ log.debug("写响应失败: {}", e.getMessage());
+ } catch (Exception e) {
+ log.error("处理请求时发生未预期错误", e);
+ }
+ }
+
+ static HttpRequestModel toRequestModel(HttpExchange exchange) {
+ String rawPath = exchange.getRequestURI().getRawPath();
+ String decodedPath = decodePath(rawPath);
+ return new HttpRequestModel(
+ exchange.getRequestMethod(),
+ rawPath,
+ decodedPath,
+ parseQuery(exchange.getRequestURI().getRawQuery()),
+ exchange.getRequestHeaders().getFirst("Range"),
+ remoteAddress(exchange));
+ }
+
+ /** 对端 IP；取不到时返回 null，交由上层按既有策略处理。 */
+ static String remoteAddress(HttpExchange exchange) {
+ InetSocketAddress remote = exchange.getRemoteAddress();
+ return remote == null || remote.getAddress() == null ? null : remote.getAddress().getHostAddress();
+ }
+
+ /** 解析查询串；值内的 {@code =} 不得截断，重复键取最后一次。 */
+ public static Map
+ *
+ *
+ *