Integrate primary-first subscriptions and durable event replay into storage node

SUBSCRIPTION_GATEWAY.md

@@ -0,0 +1,45 @@
+# storageNode 内置订阅网关
+
+8889 的 BackupSubServer 直接承载网关；无需独立进程或端口。空的
+`SubscriptionPrimaryUrl` 保留旧快照模式；生产设 `https://lionwebsite.xyz`，
+`SubscriptionEventDir` 默认 `/root/gallery/storageNode/subscription-events`，
+共享 `SubscriptionSyncSecret`，启用网关且无密钥会拒绝启动。
+
+Nginx `/sub/` 始终代理本机8889，并覆盖 `X-Real-IP $remote_addr`，保留原始
+User-Agent/Range。只有 loopback 对端可以提供 X-Real-IP；公网直连8889
+忽略所有代理来源 IP，X-Forwarded-For 始终不使用。
+
+每个有效 v2/cat GET/HEAD 优先请求固定主站 origin，连接1秒、整次含完整正文3秒；
+不传外部认证头、不跟随重定向，正文最大16MiB。网络错误、超时、5xx及3xx转备用快照；
+400/403/404等其他状态透传，不使用旧缓存。GET/HEAD、Range和内容类型/范围头保留。
+失败后熔断10秒，期间只读本地，到期只允许一个探测，成功后恢复逐请求访问主站。
+
+每次请求建立UUID事件，字段为eventId/keyHash/client/clientIp/userAgent/occurredAt。
+只存公开key的SHA256，不保存公开key、正文或密钥。UA保留合法Latin-1原始前缀，去除控制字符、非Latin-1替换为?，最大2048字符且UTF8<=2048bytes，
+并在实际base64url事件头超过4096字符时继续截断；发主站UA与事件一致。
+GET和HEAD均携带签名事件头；HEAD不会记日志。
+
+签名请求头为 X-Subscription-Event（base64url无padding紧凑JSON）、
+X-Subscription-Timestamp（epoch秒）、X-Subscription-Signature（HMAC-SHA256 hex）。
+签名输入 `v1\n<timestamp>\n<sha256(raw JSON bytes)>`，UTF8。
+主站已记但响应超时，也用原eventId回传，由主站receipt去重。
+
+备用快照200/206 GET必须先写事件文件并fsync/原子rename/目录fsync才返回；
+落盘失败或队列超限返回503，HEAD/404/416/503不写成功事件。
+目录700，文件600。上限100000条/256MiB，保留16MiB磁盘余量；达到限制拒绝新成功更新，
+不会静默丢弃旧记录。启动恢复所有json，元数据损坏会拒绝网关初始化，需先保留文件并排查。
+
+后台单线程每10秒最多回传1批，100条且原始body<=256KiB，POST
+`/internal/subscription-events`，body `{"events":[...]}`。签名使用同一规则对原始body计算，
+不含事件header。只清理本批 `{"acknowledged":["<eventId>"]}` 明确确认的事件，
+并同步目录。网络/5xx/非法JSON/未确认/未知ID均保留原ID重试。上传不持有熔断器锁。
+进程关闭会释放HTTP client和后台线程；Main shutdown hook管理完整服务生命周期。
+
+`/health/subscription` 保持现有快照字段；`/health/subscription-gateway`
+另报pendingCount/oldestAgeMillis/circuit/lastUploadAt/lastUploadState，无用户信息。
+
+验证必须在部署手册指定的构建VM运行，不能在工作机执行Java编译/测试：
+`mvn -B test`。`SubscriptionGatewayTest` 使用真实本地HTTP主站和节点入口覆盖签名、
+主站优先、fallback/持久化、ack/重启/同ID、HEAD、404、Range、timeout、熔断/单探测、
+后台上传隔离、来源IP、路径编码、磁盘错误、批次字节上限、redirect和不完整/超大正文。
+新增持久化协议全用Jackson树节点，不新增反射DTO。

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

@@ -21,6 +21,9 @@ public class Config {
     public static int subscriptionHttpPort;
     public static int subscriptionHttpWorkers;
     public static int subscriptionSocketTimeoutMs;
+    /** 空值保留快照直读模式；设置后启用进程内主站优先网关。 */
+    public static String subscriptionPrimaryUrl = "";
+    public static String subscriptionEventDir = "/root/gallery/storageNode/subscription-events";
 
     /** 下载服务的管理员口令；为空表示 8888 的管理员直取文件能力关闭。 */
     public static String adminDownloadCode = "";
@@ -76,6 +79,10 @@ public class Config {
         subscriptionHttpPort = Integer.parseInt(value(prop, "SubscriptionHttpPort", "8889"));
         subscriptionHttpWorkers = Integer.parseInt(value(prop, "SubscriptionHttpWorkers", "4"));
         subscriptionSocketTimeoutMs = Integer.parseInt(value(prop, "SubscriptionSocketTimeoutMs", "10000"));
+        subscriptionPrimaryUrl = value(prop, "SubscriptionPrimaryUrl", "");
+        subscriptionEventDir = value(prop, "SubscriptionEventDir", "/root/gallery/storageNode/subscription-events");
+        if (!subscriptionPrimaryUrl.isBlank() && subscriptionSyncSecret.isBlank())
+            throw new IllegalStateException("启用订阅网关时必须配置 SUBSCRIPTION_SYNC_SECRET");
         // 口令不再写死在源码里：优先取环境变量，其次取配置文件，都为空即关闭该能力。
         adminDownloadCode = System.getenv().getOrDefault("STORAGE_DOWNLOAD_ADMIN_CODE",
                 value(prop, "AdminDownloadCode", ""));

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

@@ -7,6 +7,7 @@ import lion.Http.HttpRequestHandler;
 import lion.Http.HttpRequestModel;
 import lion.Http.HttpResponseModel;
 import lion.Service.SubscriptionSnapshotStore;
+import lion.Service.SubscriptionGateway;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
@@ -29,7 +30,7 @@ import java.util.Objects;
  * <p>{@link #run()} 保留为旧入口（{@code Main} 用它起线程），内部改为阻塞等待，直到线程被中断。
  */
 @Slf4j
-public final class BackupSubServer implements Runnable {
+public final class BackupSubServer implements Runnable, AutoCloseable {
 
     /** 公开 Key 的合法长度下限，过短直接按不存在处理，避免无谓查表。 */
     private static final int MIN_KEY_LENGTH = 6;
@@ -41,12 +42,20 @@ public final class BackupSubServer implements Runnable {
     private final SubscriptionSnapshotStore snapshotStore;
     private final int port;
     private final int workerCount;
+    private final SubscriptionGateway gateway;
     private volatile HttpFileServer server;
 
     public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount) {
+        this(snapshotStore, port, workerCount, null);
+    }
+
+    /** gateway 由同一宿主管理生命周期；null 保持历史快照模式。 */
+    public BackupSubServer(SubscriptionSnapshotStore snapshotStore, int port, int workerCount,
+                           SubscriptionGateway gateway) {
         this.snapshotStore = Objects.requireNonNull(snapshotStore);
         this.port = port;
         this.workerCount = Math.max(1, workerCount);
+        this.gateway = gateway;
     }
 
     /** 旧入口：绑定端口、开始接收请求，并阻塞到线程被中断。 */
@@ -66,17 +75,25 @@ public final class BackupSubServer implements Runnable {
             running.close();
         } catch (IOException e) {
             log.error("备机订阅服务停止: {}", e.getMessage());
+        } finally {
+            close();
         }
     }
 
     /** 建好并启动服务；调用方负责 {@link HttpFileServer#close()}。 */
     public HttpFileServer startServer() throws IOException {
-        HttpFileServer started = HttpFileServer.create(port, "/", handler(snapshotStore), workerCount);
+        HttpFileServer started = HttpFileServer.create(port, "/", gateway == null ? handler(snapshotStore) : gateway, workerCount);
         started.start();
         server = started;
         return started;
     }
 
+    @Override
+    public void close() {
+        if (server != null) server.close();
+        if (gateway != null) gateway.close();
+    }
+
     /** 当前运行中的外壳，便于宿主统一关闭；未启动时为 null。 */
     public HttpFileServer server() {
         return server;
@@ -103,7 +120,7 @@ public final class BackupSubServer implements Runnable {
             }
         }
 
-        String[] segments = path.split("/");
+        String[] segments = path.split("/", -1);
         // "" / "sub" / <v2|cat> / <key>
         if (segments.length != 4 || !"sub".equals(segments[1])
                 || !(V2_CLIENT.equals(segments[2]) || CLASH_CLIENT.equals(segments[2])))

src/main/java/lion/Http/HttpFileServer.java

@@ -142,13 +142,17 @@ public final class HttpFileServer implements AutoCloseable {
     static HttpRequestModel toRequestModel(HttpExchange exchange) {
         String rawPath = exchange.getRequestURI().getRawPath();
         String decodedPath = decodePath(rawPath);
+        Map<String, String> requestHeaders = new LinkedHashMap<>();
+        exchange.getRequestHeaders().forEach((key, values) -> {
+            if (!values.isEmpty()) requestHeaders.put(key, values.getFirst());
+        });
         return new HttpRequestModel(
                 exchange.getRequestMethod(),
                 rawPath,
                 decodedPath,
                 parseQuery(exchange.getRequestURI().getRawQuery()),
                 exchange.getRequestHeaders().getFirst("Range"),
-                remoteAddress(exchange));
+                remoteAddress(exchange), requestHeaders);
     }
 
     /** 对端 IP；取不到时返回 null，交由上层按既有策略处理。 */
@@ -218,7 +222,7 @@ public final class HttpFileServer implements AutoCloseable {
 
         if (head || bodyLength == 0) {
             // HEAD 与空正文：手动给出长度，再以 -1 结尾确保不写正文。
-            if (head)
+            if (head && !headers.containsKey("Content-Length"))
                 headers.set("Content-Length", Long.toString(bodyLength));
             exchange.sendResponseHeaders(response.status(), -1);
             return;

src/main/java/lion/Http/HttpRequestModel.java

@@ -24,6 +24,7 @@ public final class HttpRequestModel {
     private final Map<String, String> queryParameters;
     private final String rangeHeader;
     private final String clientAddress;
+    private final Map<String, String> headers;
 
     public HttpRequestModel(String method, String rawPath, String decodedPath,
                             Map<String, String> queryParameters, String rangeHeader) {
@@ -32,6 +33,12 @@ public final class HttpRequestModel {
 
     public HttpRequestModel(String method, String rawPath, String decodedPath,
                             Map<String, String> queryParameters, String rangeHeader, String clientAddress) {
+        this(method, rawPath, decodedPath, queryParameters, rangeHeader, clientAddress, Map.of());
+    }
+
+    public HttpRequestModel(String method, String rawPath, String decodedPath,
+                            Map<String, String> queryParameters, String rangeHeader, String clientAddress,
+                            Map<String, String> headers) {
         this.method = Objects.requireNonNull(method, "method");
         this.rawPath = rawPath;
         this.decodedPath = decodedPath;
@@ -40,6 +47,10 @@ public final class HttpRequestModel {
                 : Collections.unmodifiableMap(new LinkedHashMap<>(queryParameters));
         this.rangeHeader = rangeHeader;
         this.clientAddress = clientAddress;
+        Map<String, String> normalized = new LinkedHashMap<>();
+        if (headers != null)
+            headers.forEach((key, value) -> normalized.put(key.toLowerCase(java.util.Locale.ROOT), value));
+        this.headers = Collections.unmodifiableMap(normalized);
     }
 
     public String method() {
@@ -76,6 +87,11 @@ public final class HttpRequestModel {
     }
 
     /** 对端 IP；由外壳填充，直接驱动业务逻辑时可为 null。 */
+    /** 单值请求头，名称大小写不敏感。 */
+    public String header(String name) {
+        return headers.get(name.toLowerCase(java.util.Locale.ROOT));
+    }
+
     public String clientAddress() {
         return clientAddress;
     }

src/main/java/lion/Main.java

@@ -4,6 +4,8 @@ import io.netty.bootstrap.Bootstrap;
 import lion.Config.Config;
 import lion.Externel.BackupSubServer;
 import lion.Service.SubscriptionSnapshotStore;
+import lion.Service.SubscriptionGateway;
+import java.io.IOException;
 import lombok.extern.slf4j.Slf4j;
 
 import java.nio.file.Paths;
@@ -18,8 +20,17 @@ public class Main {
                 java.nio.file.Paths.get(Config.subscriptionDataDir), Config.subscriptionSyncSecret,
                 Config.subscriptionMaxStaleSeconds, Config.subscriptionMaxPayloadBytes);
         snapshotStore.load();
-        new Thread(new BackupSubServer(snapshotStore, Config.subscriptionHttpPort, Config.subscriptionHttpWorkers),
-                "subscription-backup-http").start();
+        try {
+            SubscriptionGateway gateway = Config.subscriptionPrimaryUrl.isBlank() ? null
+                    : new SubscriptionGateway(snapshotStore, Config.subscriptionPrimaryUrl,
+                    Config.subscriptionSyncSecret, Paths.get(Config.subscriptionEventDir));
+            BackupSubServer subscriptions = new BackupSubServer(snapshotStore, Config.subscriptionHttpPort,
+                    Config.subscriptionHttpWorkers, gateway);
+            Runtime.getRuntime().addShutdownHook(new Thread(subscriptions::close, "subscription-shutdown"));
+            new Thread(subscriptions, "subscription-backup-http").start();
+        } catch (IOException failure) {
+            throw new IllegalStateException("无法初始化订阅事件持久化目录", failure);
+        }
         new Thread(() -> MultiThreadedHTTPServer.main(null)).start();
         // 启用可靠任务协议：任务与待 ACK 报告持久化到独立目录，重启后恢复。
         new storageNode(snapshotStore, Paths.get(Config.taskJournalDir), true);

src/main/java/lion/Service/SubscriptionEventJournal.java

@@ -0,0 +1,135 @@
+package lion.Service;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import lion.CustomUtil;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.nio.file.StandardOpenOption;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.UUID;
+
+/** 每个成功的备用 GET 先同步落盘；只在主站明确确认后删除。无公开 Key 和订阅正文。 */
+public final class SubscriptionEventJournal {
+    private static final long MAX_BYTES = 256L * 1024 * 1024;
+    private static final int MAX_EVENTS = 100_000;
+    private static final int MAX_EVENT_BYTES = 16 * 1024;
+    private static final long RESERVE_BYTES = 16L * 1024 * 1024;
+    private final Path directory;
+    private final TreeMap<String, Pending> events = new TreeMap<>();
+    private long bytes;
+
+    public record Pending(String filename, String eventId, long occurredAt, byte[] json) { }
+
+    public SubscriptionEventJournal(Path directory) throws IOException {
+        this.directory = directory;
+        Files.createDirectories(directory);
+        if (Files.isSymbolicLink(directory)) throw new IOException("Subscription journal cannot be a symlink");
+        Files.setPosixFilePermissions(directory, PosixFilePermissions.fromString("rwx------"));
+        try (var files = Files.list(directory)) {
+            for (Path file : files.filter(p -> p.getFileName().toString().endsWith(".json")).toList()) {
+                if (!Files.isRegularFile(file, java.nio.file.LinkOption.NOFOLLOW_LINKS)
+                        || Files.size(file) > MAX_EVENT_BYTES)
+                    throw new IOException("Invalid subscription event file");
+                byte[] json = Files.readAllBytes(file);
+                JsonNode value = CustomUtil.objectMapper.readTree(json);
+                Pending pending = pending(file.getFileName().toString(), value, json);
+                Files.setPosixFilePermissions(file, PosixFilePermissions.fromString("rw-------"));
+                events.put(pending.filename(), pending);
+                bytes += json.length;
+            }
+        }
+    }
+
+    private static Pending pending(String filename, JsonNode value, byte[] json) throws IOException {
+        try {
+            String eventId = UUID.fromString(value.path("eventId").asText()).toString();
+            long occurredAt = value.path("occurredAt").asLong(-1);
+            if (!value.isObject() || !value.path("occurredAt").isIntegralNumber() || occurredAt <= 0
+                    || !filename.equals(filename(occurredAt, eventId))
+                    || !value.path("keyHash").isTextual() || !value.path("keyHash").asText().matches("[a-f0-9]{64}")
+                    || !value.path("client").isTextual() || !Set.of("v2", "cat").contains(value.path("client").asText())
+                    || !value.path("clientIp").isTextual() || value.path("clientIp").asText().length() > 45
+                    || !value.path("clientIp").asText().matches("[0-9a-fA-F:.]+")
+                    || !value.path("userAgent").isTextual() || value.path("userAgent").asText().length() > 2048)
+                throw new IllegalArgumentException("Invalid event metadata");
+            Set<String> fields = Set.of("eventId", "keyHash", "client", "clientIp", "userAgent", "occurredAt");
+            var fieldNames = value.fieldNames();
+            while (fieldNames.hasNext()) if (!fields.contains(fieldNames.next()))
+                throw new IllegalArgumentException("Unknown event field");
+            return new Pending(filename, eventId, occurredAt, json);
+        } catch (IllegalArgumentException failure) {
+            throw new IOException("Invalid subscription event metadata", failure);
+        }
+    }
+
+    private static String filename(long time, String id) {
+        return String.format(java.util.Locale.ROOT, "%019d-%s.json", time, id);
+    }
+
+    public synchronized void append(JsonNode event) throws IOException {
+        byte[] json = CustomUtil.objectMapper.writeValueAsBytes(event);
+        Pending pending = pending(filename(event.path("occurredAt").asLong(), event.path("eventId").asText()), event, json);
+        if (events.containsKey(pending.filename())) return;
+        if (json.length > MAX_EVENT_BYTES || events.size() >= MAX_EVENTS || bytes + json.length > MAX_BYTES
+                || Files.getFileStore(directory).getUsableSpace() < RESERVE_BYTES + json.length)
+            throw new IOException("Subscription event journal capacity exceeded");
+        Path target = directory.resolve(pending.filename());
+        Path temporary = directory.resolve(pending.filename() + ".tmp");
+        try {
+            try (FileChannel channel = FileChannel.open(temporary, Set.of(StandardOpenOption.CREATE_NEW,
+                    StandardOpenOption.WRITE), PosixFilePermissions.asFileAttribute(
+                    PosixFilePermissions.fromString("rw-------")))) {
+                ByteBuffer buffer = ByteBuffer.wrap(json);
+                while (buffer.hasRemaining()) channel.write(buffer);
+                channel.force(true);
+            }
+            Files.move(temporary, target, StandardCopyOption.ATOMIC_MOVE);
+            events.put(pending.filename(), pending);
+            bytes += json.length;
+            forceDirectory();
+        } finally {
+            Files.deleteIfExists(temporary);
+        }
+    }
+
+    public synchronized List<Pending> oldest(int limit) {
+        return new ArrayList<>(events.values().stream().limit(limit).toList());
+    }
+
+    /** 未确认或本批以外的 ID 绝不清理；目录同步失败时保留内存项，下轮继续幂等重试。 */
+    public synchronized void acknowledge(List<Pending> batch, Set<String> acknowledged) throws IOException {
+        List<Pending> deleted = new ArrayList<>();
+        for (Pending pending : batch) {
+            if (acknowledged.contains(pending.eventId()) && events.containsKey(pending.filename())) {
+                Files.deleteIfExists(directory.resolve(pending.filename()));
+                deleted.add(pending);
+            }
+        }
+        if (deleted.isEmpty()) return;
+        forceDirectory();
+        for (Pending pending : deleted) {
+            events.remove(pending.filename());
+            bytes -= pending.json().length;
+        }
+    }
+
+    private void forceDirectory() throws IOException {
+        try (FileChannel channel = FileChannel.open(directory, StandardOpenOption.READ)) {
+            channel.force(true);
+        }
+    }
+
+    public synchronized int size() { return events.size(); }
+    public synchronized long oldestAgeMillis() {
+        return events.isEmpty() ? 0 : Math.max(0, System.currentTimeMillis() - events.firstEntry().getValue().occurredAt());
+    }
+}

src/main/java/lion/Service/SubscriptionGateway.java

@@ -0,0 +1,338 @@
+package lion.Service;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import lion.CustomUtil;
+import lion.Externel.BackupSubServer;
+import lion.Http.HttpRequestHandler;
+import lion.Http.HttpRequestModel;
+import lion.Http.HttpResponseModel;
+import lombok.extern.slf4j.Slf4j;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.time.Duration;
+import java.util.Base64;
+import java.util.HashSet;
+import java.util.HexFormat;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.Flow;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/** storageNode 内置的主站优先订阅入口及离线访问记录回传器。 */
+@Slf4j
+public final class SubscriptionGateway implements HttpRequestHandler, AutoCloseable {
+    private static final int MAX_RESPONSE_BYTES = 16 * 1024 * 1024;
+    private static final int MAX_BATCH_BYTES = 256 * 1024;
+    private final HttpRequestHandler local;
+    private final URI origin;
+    private final byte[] secret;
+    private final SubscriptionEventJournal journal;
+    private final HttpClient client;
+    private final Duration timeout;
+    private final long cooldownMillis;
+    private final ScheduledExecutorService uploader;
+    private final Object uploadLock = new Object();
+    private long openUntil;
+    private long generation;
+    private boolean halfOpen;
+    private volatile long lastUploadAt;
+    private volatile String lastUploadState = "idle";
+    private volatile boolean closed;
+
+    public SubscriptionGateway(SubscriptionSnapshotStore store, String primaryUrl, String secret, Path eventDir)
+            throws IOException {
+        this(store, primaryUrl, secret, eventDir, Duration.ofSeconds(3), 10_000, true);
+    }
+
+    /** 可配置时间仅供真实 HTTP 测试；生产固定 1s 连接/3s 总时限/10s 熔断。 */
+    public SubscriptionGateway(SubscriptionSnapshotStore store, String primaryUrl, String secret, Path eventDir,
+                               Duration timeout, long cooldownMillis, boolean backgroundUpload) throws IOException {
+        this.local = BackupSubServer.handler(store);
+        this.origin = validateOrigin(primaryUrl);
+        if (secret == null || secret.isBlank()) throw new IllegalArgumentException("Subscription gateway secret required");
+        this.secret = secret.getBytes(StandardCharsets.UTF_8);
+        this.journal = new SubscriptionEventJournal(eventDir);
+        this.timeout = timeout;
+        this.cooldownMillis = cooldownMillis;
+        this.client = HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(1))
+                .followRedirects(HttpClient.Redirect.NEVER).build();
+        this.uploader = Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread thread = new Thread(r, "subscription-events-upload");
+            thread.setDaemon(true);
+            return thread;
+        });
+        if (backgroundUpload) uploader.scheduleWithFixedDelay(this::uploadSafely, 10, 10, TimeUnit.SECONDS);
+    }
+
+    private static URI validateOrigin(String value) {
+        URI uri = URI.create(value);
+        if (!("http".equals(uri.getScheme()) || "https".equals(uri.getScheme()))
+                || uri.getHost() == null || uri.getUserInfo() != null || uri.getQuery() != null
+                || uri.getFragment() != null || !(uri.getPath().isEmpty() || "/".equals(uri.getPath())))
+            throw new IllegalArgumentException("SubscriptionPrimaryUrl must be a fixed HTTP(S) origin");
+        return uri;
+    }
+
+    @Override
+    public HttpResponseModel handle(HttpRequestModel request) {
+        String path = request.decodedPath();
+        if ("/health/subscription-gateway".equals(path)) {
+            if (!HttpRequestModel.GET.equals(request.method()) && !request.isHead()) return HttpResponseModel.status(405);
+            try {
+                return HttpResponseModel.bytes(200, CustomUtil.objectMapper.writeValueAsBytes(status()),
+                        "application/json; charset=utf-8", null, Map.of());
+            } catch (IOException failure) {
+                return HttpResponseModel.status(500);
+            }
+        }
+        if (!HttpRequestModel.GET.equals(request.method()) && !request.isHead()) return local.handle(request);
+        String[] parts = path == null ? new String[0] : path.split("/", -1);
+        if (parts.length != 4 || !parts[0].isEmpty() || !"sub".equals(parts[1])
+                || !("v2".equals(parts[2]) || "cat".equals(parts[2]))
+                || parts[3].length() < 6 || parts[3].length() > 512)
+            return local.handle(request);
+        // 内部路径不能成为 URI 目录跳转或带控制字符的请求。
+        if (parts[3].codePoints().anyMatch(c -> Character.isISOControl(c))) return HttpResponseModel.status(404);
+        ObjectNode event;
+        try {
+            event = event(request, parts[2], parts[3]);
+        } catch (IOException failure) {
+            return HttpResponseModel.status(503);
+        }
+        Attempt attempt = acquire();
+        if (attempt != null) {
+            try {
+                HttpResponseModel response = primary(request, parts, event);
+                if (response.status() < 500 && !(response.status() >= 300 && response.status() < 400)) {
+                    recovered(attempt);
+                    return response;
+                }
+                failed(attempt);
+            } catch (Exception failure) {
+                if (failure instanceof InterruptedException) Thread.currentThread().interrupt();
+                failed(attempt);
+                // 不记录 URL/公开 key/密钥，失败原因只暴露类别。
+                log.debug("订阅主站请求失败: {}", failure.getClass().getSimpleName());
+            }
+        }
+        HttpResponseModel fallback = local.handle(request);
+        if (!request.isHead() && (fallback.status() == 200 || fallback.status() == 206)) {
+            try {
+                journal.append(event);
+            } catch (IOException failure) {
+                log.error("订阅离线记录无法持久化，拒绝本次更新: {}", failure.getClass().getSimpleName());
+                return HttpResponseModel.text(503, "text/plain", "Subscription access journal unavailable");
+            }
+        }
+        return fallback;
+    }
+
+    private ObjectNode event(HttpRequestModel request, String format, String key) throws IOException {
+        ObjectNode event = CustomUtil.objectMapper.createObjectNode();
+        event.put("eventId", UUID.randomUUID().toString());
+        event.put("keyHash", sha256(key.getBytes(StandardCharsets.UTF_8)));
+        event.put("client", format);
+        event.put("clientIp", clientIp(request));
+        String rawAgent = request.header("User-Agent");
+        if (rawAgent == null) rawAgent = "";
+        // HTTP/1 header值使用Latin-1。正常原始UA保持不变；移除控制字符，
+        // 非Latin-1字符替换为?，避免非法header导致每次请求都只能fallback。
+        StringBuilder normalized = new StringBuilder();
+        rawAgent.codePoints().filter(c -> c >= 32 && c != 127)
+                .limit(2048).forEach(c -> normalized.appendCodePoint(c <= 255 ? c : '?'));
+        String agent = normalized.toString();
+        while (agent.getBytes(StandardCharsets.UTF_8).length > 2048)
+            agent = agent.substring(0, agent.length() - 1);
+        event.put("userAgent", agent);
+        event.put("occurredAt", System.currentTimeMillis());
+        // 签名事件头的 base64url 必须 <=4096。保留能装下的原始前缀，避免切断代理对。
+        while (Base64.getUrlEncoder().withoutPadding().encodeToString(
+                CustomUtil.objectMapper.writeValueAsBytes(event)).length() > 4096) {
+            agent = agent.substring(0, agent.offsetByCodePoints(agent.length(), -1));
+            event.put("userAgent", agent);
+        }
+        return event;
+    }
+
+    public static String clientIp(HttpRequestModel request) {
+        InetAddress remote = literalAddress(request.clientAddress());
+        if (remote != null && remote.isLoopbackAddress()) {
+            InetAddress forwarded = literalAddress(request.header("X-Real-IP"));
+            if (forwarded != null) return forwarded.getHostAddress();
+        }
+        return remote == null ? "0.0.0.0" : remote.getHostAddress();
+    }
+
+    private static InetAddress literalAddress(String value) {
+        if (value == null || value.length() > 45 || !value.matches("[0-9a-fA-F:.]+")) return null;
+        if (!value.contains(":")) {
+            if (!value.matches("[0-9]{1,3}(\\.[0-9]{1,3}){3}")) return null;
+            for (String part : value.split("\\.")) if (Integer.parseInt(part) > 255) return null;
+        }
+        try { return InetAddress.getByName(value); } catch (Exception ignored) { return null; }
+    }
+
+    private HttpResponseModel primary(HttpRequestModel request, String[] parts, ObjectNode event) throws Exception {
+        URI target = endpoint("/sub/" + parts[2] + "/" + parts[3]);
+        HttpRequest.Builder builder = HttpRequest.newBuilder(target).timeout(timeout)
+                .method(request.isHead() ? "HEAD" : "GET", HttpRequest.BodyPublishers.noBody())
+                .header("User-Agent", event.path("userAgent").asText());
+        if (request.rangeHeader() != null) builder.header("Range", request.rangeHeader());
+        // HEAD 也必须签名，使主站绕过普通访问记录分支；主站与节点均不记 HEAD。
+        byte[] json = CustomUtil.objectMapper.writeValueAsBytes(event);
+        builder.header("X-Subscription-Event", Base64.getUrlEncoder().withoutPadding().encodeToString(json));
+        sign(builder, json);
+        HttpResponse<byte[]> response = send(builder.build(), MAX_RESPONSE_BYTES);
+        Map<String, String> headers = new LinkedHashMap<>();
+        for (String name : List.of("Content-Range", "Accept-Ranges"))
+            response.headers().firstValue(name).ifPresent(value -> headers.put(name, value));
+        if (request.isHead()) response.headers().firstValue("Content-Length")
+                .ifPresent(value -> headers.put("Content-Length", value));
+        return HttpResponseModel.bytes(response.statusCode(), response.body(),
+                response.headers().firstValue("Content-Type").orElse("application/octet-stream"), null, headers);
+    }
+
+    /** URI 构造器百分号编码单一路径；从不拼接用户 origin/query 或跟随 redirect。 */
+    private URI endpoint(String path) throws URISyntaxException {
+        return new URI(origin.getScheme(), null, origin.getHost(), origin.getPort(), path, null, null);
+    }
+
+    private void sign(HttpRequest.Builder builder, byte[] payload) throws Exception {
+        String timestamp = Long.toString(System.currentTimeMillis() / 1000);
+        Mac mac = Mac.getInstance("HmacSHA256");
+        mac.init(new SecretKeySpec(secret, "HmacSHA256"));
+        String input = "v1\n" + timestamp + "\n" + sha256(payload);
+        builder.header("X-Subscription-Timestamp", timestamp)
+                .header("X-Subscription-Signature", HexFormat.of().formatHex(mac.doFinal(input.getBytes(StandardCharsets.UTF_8))));
+    }
+
+    private static String sha256(byte[] bytes) {
+        try { return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes)); }
+        catch (java.security.NoSuchAlgorithmException impossible) { throw new IllegalStateException(impossible); }
+    }
+
+    /** sendAsync 的整次 future 包含正文消费，避免 ofInputStream 绕过读取时限。 */
+    private HttpResponse<byte[]> send(HttpRequest request, int maxBytes) throws Exception {
+        CompletableFuture<HttpResponse<byte[]>> future = client.sendAsync(request,
+                info -> new BoundedSubscriber(maxBytes));
+        try { return future.get(timeout.toMillis(), TimeUnit.MILLISECONDS); }
+        finally { if (!future.isDone()) future.cancel(true); }
+    }
+
+    private static final class BoundedSubscriber implements HttpResponse.BodySubscriber<byte[]> {
+        private final HttpResponse.BodySubscriber<byte[]> delegate = HttpResponse.BodySubscribers.ofByteArray();
+        private final int limit;
+        private int received;
+        private Flow.Subscription subscription;
+        BoundedSubscriber(int limit) { this.limit = limit; }
+        public CompletionStage<byte[]> getBody() { return delegate.getBody(); }
+        public void onSubscribe(Flow.Subscription value) { subscription = value; delegate.onSubscribe(value); }
+        public void onNext(List<ByteBuffer> buffers) {
+            long size = buffers.stream().mapToLong(ByteBuffer::remaining).sum();
+            if (size > limit - received) {
+                subscription.cancel();
+                delegate.onError(new IOException("Subscription HTTP response exceeds limit"));
+            } else { received += (int) size; delegate.onNext(buffers); }
+        }
+        public void onError(Throwable error) { delegate.onError(error); }
+        public void onComplete() { delegate.onComplete(); }
+    }
+
+    private record Attempt(long generation) { }
+    private synchronized Attempt acquire() {
+        if (openUntil == 0) return new Attempt(generation);
+        if (System.currentTimeMillis() < openUntil || halfOpen) return null;
+        halfOpen = true;
+        return new Attempt(generation);
+    }
+    private synchronized void failed(Attempt attempt) {
+        if (attempt.generation() != generation) return;
+        openUntil = System.currentTimeMillis() + cooldownMillis;
+        halfOpen = false;
+        generation++;
+    }
+    private synchronized void recovered(Attempt attempt) {
+        if (attempt.generation() != generation) return;
+        openUntil = 0;
+        halfOpen = false;
+    }
+
+    public Map<String, Object> status() {
+        String circuit;
+        synchronized (this) { circuit = openUntil == 0 ? "closed" : halfOpen ? "half-open" : "open"; }
+        return Map.of("pendingCount", journal.size(), "oldestAgeMillis", journal.oldestAgeMillis(),
+                "circuit", circuit, "lastUploadAt", lastUploadAt, "lastUploadState", lastUploadState);
+    }
+
+    private void uploadSafely() {
+        try { uploadOnce(); }
+        catch (Exception failure) {
+            lastUploadState = "failed";
+            log.debug("订阅访问记录回传失败: {}", failure.getClass().getSimpleName());
+        }
+    }
+
+    /** 每轮最多一批、100条、256KiB；所有未明确确认的原始 eventId 均留盘。 */
+    public void uploadOnce() throws Exception {
+        synchronized (uploadLock) {
+            if (closed) return;
+            List<SubscriptionEventJournal.Pending> candidates = journal.oldest(100);
+            if (candidates.isEmpty()) { lastUploadState = "idle"; return; }
+            ObjectNode body = CustomUtil.objectMapper.createObjectNode();
+            var array = body.putArray("events");
+            java.util.ArrayList<SubscriptionEventJournal.Pending> batch = new java.util.ArrayList<>();
+            int total = 13; // {"events":[]} + 逗号安全余量
+            for (var pending : candidates) {
+                if (total + pending.json().length + 1 > MAX_BATCH_BYTES) break;
+                array.add(CustomUtil.objectMapper.readTree(pending.json()));
+                batch.add(pending);
+                total += pending.json().length + 1;
+            }
+            if (batch.isEmpty()) throw new IOException("Subscription event exceeds upload batch limit");
+            byte[] json = CustomUtil.objectMapper.writeValueAsBytes(body);
+            HttpRequest.Builder builder = HttpRequest.newBuilder(endpoint("/internal/subscription-events"))
+                    .timeout(timeout).header("Content-Type", "application/json")
+                    .POST(HttpRequest.BodyPublishers.ofByteArray(json));
+            sign(builder, json);
+            HttpResponse<byte[]> response = send(builder.build(), 64 * 1024);
+            if (response.statusCode() != 200) throw new IOException("Subscription event upload rejected");
+            JsonNode result = CustomUtil.objectMapper.readTree(response.body());
+            if (result == null || !result.path("acknowledged").isArray()) throw new IOException("Invalid subscription event acknowledgement");
+            Set<String> ids = new HashSet<>();
+            for (JsonNode id : result.path("acknowledged")) {
+                if (!id.isTextual()) throw new IOException("Invalid subscription event acknowledgement");
+                ids.add(id.asText());
+            }
+            journal.acknowledge(batch, ids);
+            lastUploadAt = System.currentTimeMillis();
+            lastUploadState = "ok";
+        }
+    }
+
+    @Override
+    public void close() {
+        closed = true;
+        uploader.shutdownNow();
+        client.close();
+    }
+}

src/main/resources/config.properties

@@ -9,3 +9,9 @@ SubscriptionHttpWorkers=4
 SubscriptionSocketTimeoutMs=10000
 # 可靠任务协议的任务日志目录（独立于订阅数据目录，默认不接触订阅/测试生产目录）
 TaskJournalDir=/root/gallery/storageNode/task-journal
+
+# 主站优先网关（与订阅服务共用8889；空值关闭，保留快照直读）
+# 生产设置 https://lionwebsite.xyz，仅允许固定origin，不含路径/认证信息。
+SubscriptionPrimaryUrl=
+# 离线成功GET的耐久队列，独立于snapshot/task-journal，权限700/600。
+SubscriptionEventDir=/root/gallery/storageNode/subscription-events

src/test/java/lion/ConfigTest.java

@@ -31,6 +31,8 @@ class ConfigTest {
         assertEquals(8889, Config.subscriptionHttpPort);
         assertEquals(4, Config.subscriptionHttpWorkers);
         assertEquals(10000, Config.subscriptionSocketTimeoutMs);
+        assertEquals("", Config.subscriptionPrimaryUrl);
+        assertEquals("/root/gallery/storageNode/subscription-events", Config.subscriptionEventDir);
     }
 
     @Test
@@ -94,6 +96,18 @@ class ConfigTest {
         assertEquals("/root/gallery", Config.adminDownloadRoot);
     }
 
+    @Test
+    void gatewayConfigurationRequiresSecretAndHasIndependentEventDirectory(@TempDir Path root) throws Exception {
+        Path file = Files.writeString(root.resolve("gateway.properties"),
+                "SubscriptionPrimaryUrl=https://lionwebsite.xyz\nSubscriptionEventDir=/tmp/events\n");
+        assertThrows(IllegalStateException.class, () -> Config.loadConfig(file.toString()));
+        Files.writeString(file, "SubscriptionPrimaryUrl=https://lionwebsite.xyz\n"
+                + "SubscriptionSyncSecret=gateway-secret\nSubscriptionEventDir=/tmp/events\n");
+        Config.loadConfig(file.toString());
+        assertEquals("https://lionwebsite.xyz", Config.subscriptionPrimaryUrl);
+        assertEquals("/tmp/events", Config.subscriptionEventDir);
+    }
+
     /**
      * 启用订阅同步却没有密钥时必须拒绝启动。
      * 否则备机会以空密钥运行，签名校验形同虚设。

src/test/java/lion/Service/SubscriptionGatewayTest.java

@@ -0,0 +1,469 @@
+package lion.Service;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import lion.CustomUtil;
+import lion.Externel.BackupSubServer;
+import lion.Http.HttpRequestModel;
+import lion.Http.HttpResponseModel;
+import lion.Message.Main.SubscriptionAccountSnapshot;
+import lion.Message.Main.SubscriptionBindingSnapshot;
+import lion.Message.Main.SubscriptionSnapshotMessage;
+import lion.Message.Main.SubscriptionSnapshotPayload;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.zip.GZIPOutputStream;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class SubscriptionGatewayTest {
+    private static final String SECRET = "gateway-test-secret";
+    private static final String KEY = "public-key-abcdef";
+    private static final String LOCAL = "last synced snapshot";
+    private static final HttpClient HTTP = HttpClient.newHttpClient();
+    @TempDir Path root;
+
+    @Test
+    void primaryPreferredAndSignedWithoutForwardingExternalAuth() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            HttpResponse<String> response = h.request("GET", "User-Agent", "Clash/1.2", "X-Real-IP", "198.51.100.9",
+                    "Authorization", "Bearer never-forward", "X-Subscription-Event", "attacker");
+            assertEquals(200, response.statusCode());
+            assertEquals("fresh primary", response.body());
+            assertEquals(0, h.pending());
+            JsonNode event = h.lastEvent.get();
+            assertEquals(hash(KEY.getBytes(StandardCharsets.UTF_8)), event.path("keyHash").asText());
+            assertEquals("Clash/1.2", event.path("userAgent").asText());
+            assertEquals("198.51.100.9", event.path("clientIp").asText());
+            assertEquals("v2", event.path("client").asText());
+            assertNull(h.authorization.get());
+            assertTrue(h.signatureValid);
+            assertFalse(new String(h.eventBytes.get(), StandardCharsets.UTF_8).contains(KEY));
+        }
+    }
+
+    @Test
+    void fallbackDurablyRecordsAndUploadAcknowledgementClears() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            h.primaryStatus.set(503);
+            assertEquals(LOCAL, h.request("GET", "User-Agent", "real-agent").body());
+            assertEquals(1, h.pending());
+            SubscriptionEventJournal recovered = new SubscriptionEventJournal(h.eventDir);
+            assertEquals(1, recovered.size(), "事件必须可以从磁盘恢复");
+            String eventId = recovered.oldest(1).getFirst().eventId();
+            h.gateway.uploadOnce();
+            assertEquals(0, h.pending());
+            assertEquals(eventId, h.uploaded.get().path("events").get(0).path("eventId").asText());
+            assertEquals(0, new SubscriptionEventJournal(h.eventDir).size());
+            assertTrue(h.signatureValid);
+        }
+    }
+
+    @Test
+    void badUnknownAndMissingAcknowledgementsRetainStableIdsAcrossRestart() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            h.primaryStatus.set(503);
+            h.request("GET");
+            String id = new SubscriptionEventJournal(h.eventDir).oldest(1).getFirst().eventId();
+            h.ackMode.set(1); // 未知ID
+            h.gateway.uploadOnce();
+            assertEquals(1, h.pending());
+            h.ackMode.set(2); // 非法JSON
+            assertThrows(Exception.class, h.gateway::uploadOnce);
+            assertEquals(1, h.pending());
+            h.ackMode.set(3); // 没有ack字段
+            assertThrows(Exception.class, h.gateway::uploadOnce);
+            assertEquals(1, h.pending());
+            h.gateway.close();
+            try (SubscriptionGateway restarted = new SubscriptionGateway(h.store, h.origin, SECRET, h.eventDir,
+                    Duration.ofMillis(500), 100, false)) {
+                assertEquals(1, ((Number) restarted.status().get("pendingCount")).intValue());
+                h.ackMode.set(0);
+                restarted.uploadOnce();
+                assertEquals(id, h.uploaded.get().path("events").get(0).path("eventId").asText());
+                assertEquals(0, ((Number) restarted.status().get("pendingCount")).intValue());
+            }
+        }
+    }
+
+    @Test
+    void headAndNotFoundDoNotProduceSuccessfulRecordsOrFallback() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            h.primaryStatus.set(404);
+            HttpResponse<String> missing = h.request("GET");
+            assertEquals(404, missing.statusCode());
+            assertEquals("not found", missing.body());
+            assertEquals(0, h.pending());
+            h.primaryStatus.set(503);
+            assertEquals(200, h.request("HEAD").statusCode());
+            assertEquals(0, h.pending());
+            assertEquals(1, h.primaryGets.get(), "HEAD 不得被记成GET");
+        }
+    }
+
+    @Test
+    void primaryHeadAndRangeHeadersSurviveGateway() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            HttpResponse<String> head = h.request("HEAD");
+            assertEquals("13", head.headers().firstValue("Content-Length").orElseThrow());
+            assertEquals("", head.body());
+            assertEquals(0, h.pending());
+            assertNotNull(h.lastEvent.get(), "HEAD 必须附签名，主站才不会按普通请求记录");
+            assertTrue(h.signatureValid);
+            HttpResponse<String> range = h.request("GET", "Range", "bytes=0-4");
+            assertEquals(206, range.statusCode());
+            assertEquals("fresh", range.body());
+            assertEquals("bytes 0-4/13", range.headers().firstValue("Content-Range").orElseThrow());
+            assertEquals("bytes", range.headers().firstValue("Accept-Ranges").orElseThrow());
+            assertEquals("bytes=0-4", h.lastRange.get());
+        }
+    }
+
+    @Test
+    void primaryRecordedBeforeTimeoutReplaysSameEventId() throws Exception {
+        try (Harness h = new Harness(root, 200, 100)) {
+            h.primaryDelay.set(650);
+            long start = System.nanoTime();
+            HttpResponse<String> response = h.request("GET");
+            assertEquals(LOCAL, response.body());
+            assertTrue(Duration.ofNanos(System.nanoTime() - start).toMillis() < 1500);
+            String id = h.lastEvent.get().path("eventId").asText();
+            assertEquals(id, new SubscriptionEventJournal(h.eventDir).oldest(1).getFirst().eventId());
+            h.gateway.uploadOnce();
+            assertEquals(id, h.uploaded.get().path("events").get(0).path("eventId").asText());
+            assertEquals(0, h.pending());
+        }
+    }
+
+    @Test
+    void circuitSkipsFailedOriginThenRecoversToPrimary() throws Exception {
+        try (Harness h = new Harness(root, 500, 150)) {
+            h.primaryStatus.set(503);
+            assertEquals(LOCAL, h.request("GET").body());
+            assertEquals(LOCAL, h.request("GET").body());
+            assertEquals(1, h.primaryGets.get());
+            assertEquals("open", h.gateway.status().get("circuit"));
+            h.primaryStatus.set(200);
+            Thread.sleep(180);
+            assertEquals("fresh primary", h.request("GET").body());
+            assertEquals("closed", h.gateway.status().get("circuit"));
+            assertEquals("fresh primary", h.request("GET").body());
+            assertEquals(3, h.primaryGets.get());
+        }
+    }
+
+    @Test
+    void halfOpenAllowsOnlyOneProbeAndBackgroundUploadDoesNotBlockRequests() throws Exception {
+        try (Harness h = new Harness(root, 1000, 100)) {
+            h.primaryStatus.set(503);
+            h.request("GET");
+            Thread.sleep(130);
+            h.primaryStatus.set(200);
+            h.probeGate = new CountDownLatch(1);
+            CompletableFuture<HttpResponse<String>> probe = CompletableFuture.supplyAsync(() -> {
+                try { return h.request("GET"); } catch (Exception e) { throw new RuntimeException(e); }
+            });
+            assertTrue(h.probeEntered.await(1, TimeUnit.SECONDS));
+            assertEquals(LOCAL, h.request("GET").body());
+            assertEquals(2, h.primaryGets.get());
+            h.probeGate.countDown();
+            assertEquals("fresh primary", probe.get(2, TimeUnit.SECONDS).body());
+
+            h.uploadDelay.set(700);
+            CompletableFuture<Void> upload = CompletableFuture.runAsync(() -> {
+                try { h.gateway.uploadOnce(); } catch (Exception e) { throw new RuntimeException(e); }
+            });
+            assertTrue(h.uploadEntered.await(1, TimeUnit.SECONDS));
+            long start = System.nanoTime();
+            assertEquals("fresh primary", h.request("GET").body());
+            assertTrue(Duration.ofNanos(System.nanoTime() - start).toMillis() < 500,
+                    "上传不得持有请求路径使用的熔断器锁");
+            upload.get(2, TimeUnit.SECONDS);
+        }
+    }
+
+    @Test
+    void diskFailureReturns503InsteadOfUnrecordedSuccess() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            Files.delete(h.eventDir);
+            Files.writeString(h.eventDir, "not a directory");
+            h.primaryStatus.set(503);
+            assertEquals(503, h.request("GET").statusCode());
+            assertEquals(0, h.pending());
+        }
+    }
+
+    @Test
+    void onlyLoopbackCanSupplyRealIpAndXffIsAlwaysIgnored() {
+        assertEquals("203.0.113.12", SubscriptionGateway.clientIp(model("GET", KEY, "203.0.113.12",
+                Map.of("X-Real-IP", "198.51.100.8", "X-Forwarded-For", "198.51.100.7"))));
+        assertEquals("198.51.100.8", SubscriptionGateway.clientIp(model("GET", KEY, "127.0.0.1",
+                Map.of("X-Real-IP", "198.51.100.8", "X-Forwarded-For", "198.51.100.7"))));
+        assertEquals("127.0.0.1", SubscriptionGateway.clientIp(model("GET", KEY, "127.0.0.1",
+                Map.of("X-Real-IP", "attacker.example", "X-Forwarded-For", "198.51.100.7"))));
+    }
+
+    @Test
+    void encodedKeyNeverChangesOriginAndInvalidPathsNeverContactPrimary() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            String key = "safe+key?next=%23#fragment";
+            assertEquals(200, h.gateway.handle(model("GET", key, "203.0.113.1", Map.of())).status());
+            assertEquals("/sub/v2/" + key, h.lastPath.get());
+            for (String path : List.of("/sub/v2/abc", "/sub/v2/abcdef/", "/sub/v2/abc/def", "/other/abcdef")) {
+                assertEquals(404, h.gateway.handle(new HttpRequestModel("GET", path, path, Map.of(), null)).status());
+            }
+            assertEquals(1, h.primaryGets.get());
+            assertThrows(IllegalArgumentException.class, () -> new SubscriptionGateway(h.store,
+                    h.origin + "/arbitrary/path", SECRET, root.resolve("bad")));
+        }
+    }
+
+    @Test
+    void uploadBatchRemainsUnderBodyLimitWithEscapedAgents() throws Exception {
+        try (Harness h = new Harness(root, 500, 10000)) {
+            h.primaryStatus.set(503);
+            String agent = "\\".repeat(2048);
+            for (int i = 0; i < 100; i++) {
+                assertEquals(200, h.gateway.handle(model("GET", KEY, "203.0.113.9", Map.of("User-Agent", agent))).status());
+            }
+            assertEquals(100, h.pending());
+            h.gateway.uploadOnce();
+            assertTrue(h.uploadBodyBytes.get() <= 256 * 1024);
+            assertTrue(h.pending() > 0, "100个大UA必须拆批");
+            assertTrue(h.pending() < 100);
+            for (var pending : new SubscriptionEventJournal(h.eventDir).oldest(100)) {
+                assertTrue(Base64.getUrlEncoder().withoutPadding().encodeToString(pending.json()).length() <= 4096);
+            }
+        }
+    }
+
+    @Test
+    void userAgentNormalizationKeepsHttpHeadersValidAndWithinBudget() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            assertEquals(200, h.gateway.handle(model("GET", KEY, "203.0.113.1",
+                    Map.of("User-Agent", "猫😀\r\nAgent\t"))).status());
+            assertEquals("??Agent", h.lastEvent.get().path("userAgent").asText());
+            assertEquals(0, h.pending(), "规范后应正常访问主站而非仅fallback");
+            assertEquals(200, h.gateway.handle(model("GET", KEY, "203.0.113.1",
+                    Map.of("User-Agent", "é".repeat(2048)))).status());
+            assertTrue(h.lastEvent.get().path("userAgent").asText().getBytes(StandardCharsets.UTF_8).length <= 2048);
+            assertTrue(h.eventHeaderLength.get() <= 4096);
+        }
+    }
+
+    @Test
+    void primaryRedirectIsNotFollowedAndFallsBack() throws Exception {
+        try (Harness h = new Harness(root, 500, 100)) {
+            h.primaryStatus.set(302);
+            assertEquals(LOCAL, h.request("GET").body());
+            assertEquals(1, h.primaryGets.get());
+            assertEquals(1, h.pending());
+        }
+    }
+
+    @Test
+    void oversizedOrIncompletePrimaryBodiesFallBack() throws Exception {
+        try (Harness h = new Harness(root, 1000, 1)) {
+            h.primaryStatus.set(601); // 声明较长正文，只写一个字节
+            assertEquals(LOCAL, h.request("GET").body());
+            Thread.sleep(10);
+            h.primaryStatus.set(602); // 正文超过16MiB
+            assertEquals(LOCAL, h.request("GET").body());
+            assertEquals(2, h.pending());
+        }
+    }
+
+    private static HttpRequestModel model(String method, String key, String peer, Map<String, String> headers) {
+        String path = "/sub/v2/" + key;
+        return new HttpRequestModel(method, path, path, Map.of(), null, peer, headers);
+    }
+
+    private static final class Harness implements AutoCloseable {
+        final HttpServer primary;
+        final java.util.concurrent.ExecutorService executor = Executors.newCachedThreadPool();
+        final SubscriptionSnapshotStore store;
+        final SubscriptionGateway gateway;
+        final BackupSubServer backup;
+        final Path eventDir;
+        final String origin;
+        final AtomicInteger primaryStatus = new AtomicInteger(200);
+        final AtomicInteger primaryGets = new AtomicInteger();
+        final AtomicInteger primaryDelay = new AtomicInteger();
+        final AtomicInteger uploadDelay = new AtomicInteger();
+        final AtomicInteger ackMode = new AtomicInteger();
+        final AtomicInteger uploadBodyBytes = new AtomicInteger();
+        final AtomicInteger eventHeaderLength = new AtomicInteger();
+        final AtomicReference<JsonNode> lastEvent = new AtomicReference<>();
+        final AtomicReference<JsonNode> uploaded = new AtomicReference<>();
+        final AtomicReference<byte[]> eventBytes = new AtomicReference<>();
+        final AtomicReference<String> authorization = new AtomicReference<>();
+        final AtomicReference<String> lastRange = new AtomicReference<>();
+        final AtomicReference<String> lastPath = new AtomicReference<>();
+        final CountDownLatch probeEntered = new CountDownLatch(1);
+        final CountDownLatch uploadEntered = new CountDownLatch(1);
+        volatile CountDownLatch probeGate;
+        volatile boolean signatureValid;
+
+        Harness(Path root, int timeoutMs, long cooldownMs) throws Exception {
+            eventDir = root.resolve("events");
+            store = readyStore(root.resolve("snapshot"));
+            primary = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 16);
+            primary.setExecutor(executor);
+            primary.createContext("/", this::serve);
+            primary.start();
+            origin = "http://127.0.0.1:" + primary.getAddress().getPort();
+            gateway = new SubscriptionGateway(store, origin, SECRET, eventDir,
+                    Duration.ofMillis(timeoutMs), cooldownMs, false);
+            backup = new BackupSubServer(store, 0, 4, gateway);
+            backup.startServer();
+        }
+
+        void serve(HttpExchange exchange) {
+            try (exchange) {
+                if (exchange.getRequestURI().getPath().equals("/internal/subscription-events")) {
+                    byte[] bytes = exchange.getRequestBody().readAllBytes();
+                    uploadBodyBytes.set(bytes.length);
+                    signatureValid = signatureValid(exchange, bytes);
+                    JsonNode value = CustomUtil.objectMapper.readTree(bytes);
+                    uploaded.set(value);
+                    uploadEntered.countDown();
+                    Thread.sleep(uploadDelay.get());
+                    String body;
+                    if (ackMode.get() == 1) body = "{\"acknowledged\":[\"unknown-id\"]}";
+                    else if (ackMode.get() == 2) body = "invalid-json";
+                    else if (ackMode.get() == 3) body = "{}";
+                    else {
+                        List<String> ids = new ArrayList<>();
+                        value.path("events").forEach(event -> ids.add(event.path("eventId").asText()));
+                        body = CustomUtil.objectMapper.writeValueAsString(Map.of("acknowledged", ids));
+                    }
+                    respond(exchange, 200, body);
+                    return;
+                }
+                lastPath.set(exchange.getRequestURI().getPath());
+                boolean head = "HEAD".equals(exchange.getRequestMethod());
+                if (!head) primaryGets.incrementAndGet();
+                authorization.set(exchange.getRequestHeaders().getFirst("Authorization"));
+                lastRange.set(exchange.getRequestHeaders().getFirst("Range"));
+                String encoded = exchange.getRequestHeaders().getFirst("X-Subscription-Event");
+                if (encoded != null) {
+                    eventHeaderLength.set(encoded.length());
+                    byte[] bytes = Base64.getUrlDecoder().decode(encoded);
+                    eventBytes.set(bytes);
+                    lastEvent.set(CustomUtil.objectMapper.readTree(bytes));
+                    signatureValid = signatureValid(exchange, bytes);
+                }
+                if (probeGate != null) { probeEntered.countDown(); probeGate.await(2, TimeUnit.SECONDS); }
+                Thread.sleep(primaryDelay.get());
+                int status = primaryStatus.get();
+                if (status == 601) {
+                    exchange.sendResponseHeaders(200, 100);
+                    exchange.getResponseBody().write('a');
+                    return;
+                }
+                if (status == 602) {
+                    exchange.sendResponseHeaders(200, 16 * 1024 * 1024 + 1);
+                    byte[] chunk = new byte[64 * 1024];
+                    for (int i = 0; i < 256; i++) exchange.getResponseBody().write(chunk);
+                    exchange.getResponseBody().write('z');
+                    return;
+                }
+                if (status == 302) exchange.getResponseHeaders().set("Location", origin + "/forbidden-redirect");
+                exchange.getResponseHeaders().set("Content-Type", "text/plain");
+                exchange.getResponseHeaders().set("Accept-Ranges", "bytes");
+                if (head) {
+                    exchange.getResponseHeaders().set("Content-Length", "13");
+                    exchange.sendResponseHeaders(status, -1);
+                } else if (status == 200 && lastRange.get() != null) {
+                    exchange.getResponseHeaders().set("Content-Range", "bytes 0-4/13");
+                    respond(exchange, 206, "fresh");
+                } else respond(exchange, status, status == 404 ? "not found" : "fresh primary");
+            } catch (Exception ignored) { /* 断连/短正文/超时正是测试用例 */ }
+        }
+
+        HttpResponse<String> request(String method, String... headers) throws Exception {
+            var builder = HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + backup.server().port() + "/sub/v2/" + KEY))
+                    .timeout(Duration.ofSeconds(5)).method(method, HttpRequest.BodyPublishers.noBody());
+            if (headers.length > 0) builder.headers(headers);
+            return HTTP.send(builder.build(), HttpResponse.BodyHandlers.ofString());
+        }
+        int pending() { return ((Number) gateway.status().get("pendingCount")).intValue(); }
+        @Override public void close() { backup.close(); primary.stop(0); executor.shutdownNow(); }
+    }
+
+    private static void respond(HttpExchange exchange, int code, String text) throws Exception {
+        byte[] bytes = text.getBytes(StandardCharsets.UTF_8);
+        exchange.sendResponseHeaders(code, bytes.length);
+        exchange.getResponseBody().write(bytes);
+    }
+    private static boolean signatureValid(HttpExchange exchange, byte[] bytes) throws Exception {
+        String timestamp = exchange.getRequestHeaders().getFirst("X-Subscription-Timestamp");
+        String expected = signature("v1\n" + timestamp + "\n" + hash(bytes));
+        return expected.equals(exchange.getRequestHeaders().getFirst("X-Subscription-Signature"));
+    }
+    private static String signature(String input) throws Exception {
+        Mac mac = Mac.getInstance("HmacSHA256");
+        mac.init(new SecretKeySpec(SECRET.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+        return HexFormat.of().formatHex(mac.doFinal(input.getBytes(StandardCharsets.UTF_8)));
+    }
+    private static String hash(byte[] bytes) throws Exception {
+        return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(bytes));
+    }
+    private static SubscriptionSnapshotStore readyStore(Path path) throws Exception {
+        var account = new SubscriptionAccountSnapshot();
+        account.setAccountId(1);
+        account.setEnabled(true);
+        byte[] content = LOCAL.getBytes(StandardCharsets.UTF_8);
+        account.setV2ContentBase64(Base64.getEncoder().encodeToString(content));
+        account.setV2Sha256(hash(content));
+        account.setClashContentBase64(Base64.getEncoder().encodeToString(content));
+        account.setClashSha256(hash(content));
+        var binding = new SubscriptionBindingSnapshot();
+        binding.setAccountId(1);
+        binding.setPublicKeySha256(hash(KEY.getBytes(StandardCharsets.UTF_8)));
+        var payload = new SubscriptionSnapshotPayload();
+        payload.setSchemaVersion(1);
+        payload.setAccounts(List.of(account));
+        payload.setBindings(List.of(binding));
+        byte[] json = CustomUtil.objectMapper.writeValueAsBytes(payload);
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        try (GZIPOutputStream gzip = new GZIPOutputStream(out)) { gzip.write(json); }
+        byte[] zipped = out.toByteArray();
+        var message = new SubscriptionSnapshotMessage();
+        message.setSchemaVersion(1);
+        message.setRevision(hash(json));
+        message.setGeneratedAt(System.currentTimeMillis());
+        message.setPayloadBase64(Base64.getEncoder().encodeToString(zipped));
+        message.setPayloadSha256(hash(zipped));
+        message.setSignature(signature("1\n" + message.getRevision() + "\n" + message.getGeneratedAt() + "\n" + message.getPayloadSha256()));
+        var store = new SubscriptionSnapshotStore(path, SECRET, 3600, 1024 * 1024);
+        assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code());
+        return store;
+    }
+}