修复主站通道引用被误清空导致任务状态永久卡住

问题：主站重连期间节点先后认证多条通道，引用被最后一条覆盖；那条通道断开后引用被清空，而更早建立、仍然可用的通道继续发送可用性探测。节点因引用为空不再上报任何任务状态，主站又能在旧通道上收到探活响应，双方都判定连接正常。于是未完成任务永久停在「已提交」，只有人工触发重连或重启主站才会恢复。现场证据：gid 4203383 于 19:31 创建、19:32 归档完成，节点此后每 5 秒扫到 status=4，却再未打印过「任务状态发送完成」；主站侧 17:23 之后一条「下载进度」都没有。20:28/20:29 两次收到任务下发时节点判定 server 为空而静默丢弃。修复： - 新增 PrimaryChannelTracker 统一管理主站通道引用，并保留全部已认证通道。首选通道失效时立即回退到其它仍可用的已认证通道；引用被清空时，已认证通道上的下一条消息即可恢复它。认领口径严格等于主站实际会发的消息类型（身份/任务下发/ 删除/探活/订阅快照），备机身份与节点自身出站类型都不算证据。 - 新增 RekickPolicy：有待上报任务却无可用通道时，以 30 秒为限流间隔主动重新唤起主站，不再干等最长半小时的探测周期。阻塞式探测投递到独立线程，避免占用 5 秒调度线程拖住下载扫描与压缩。 - 通道不可用时的告警与重连判断放在 downloadCheck 之前，但不提前返回，本地下载与压缩必须继续推进；否则正在下载、进度无变化的任务会走「无需上报」的提前返回路径，通道失联后既不告警也不重连，正是历史事故的成因。验证：新增 13 项测试（通道引用自愈 6、事故场景复现 3、重连限流 4），节点侧合计 28 项全绿。已部署到存储节点并与主站重新握手，双向报文正常，未完成任务数归零。
2026-09-21 22:17:14 +08:00
parent 6a97e403ba
commit 77f817d402
7 changed files with 503 additions and 22 deletions
@@ -45,13 +45,15 @@ storageNode/
    │   │       │       └── ResponseMessage.java         # 通用响应（type=0）
    │   │       └── Service/
    │   │           ├── DeleteService.java               # 删除画廊目录
-    │   │           └── DownloadCheckService.java        # 下载监控与压缩服务
+    │   │           ├── DownloadCheckService.java        # 下载监控与压缩服务
+    │   │           ├── PrimaryChannelTracker.java        # 主站通道引用登记与自愈
+    │   │           └── RekickPolicy.java                # 主动重新唤起主站的限流判定
    │   └── resources/
    │       ├── config.properties                        # DouNai 订阅地址配置
    │       ├── simplelogger.properties                  # SLF4J 日志配置（输出到 run.out）
    │       └── reflect-config.json                      # GraalVM 反射配置（Jackson 序列化）
    └── test/
-        └── java/                                        # 订阅快照与下载/压缩恢复测试
+        └── java/                                        # 订阅快照、下载/压缩恢复与主站通道自愈测试
 ```

 ---
@@ -130,6 +132,17 @@ storageNode/
 - 按名称删除画廊目录
 - 失败时返回 `ErrorCode.IO_ERROR` 或 `ErrorCode.FILE_NOT_FOUND`

+### PrimaryChannelTracker
+- 登记「哪条通道是主站」，供任务状态上报与心跳使用
+- 只有主站会发的消息类型（身份/任务下发/探活/订阅快照）才可用于认领引用；
+  备机的身份消息与节点自己的出站类型都不算证据
+- 首选通道断开时立即回退到其它仍可用的已认证通道；引用被清空时，
+  已认证通道上的下一条消息即可恢复它，避免任务状态静默停止上报
+
+### RekickPolicy
+- 判定「有待上报任务、却无可用通道」时是否应主动重新唤起主站
+- 以 30 秒为最小间隔限流，既能快速自愈，又不会在主站确实离线时形成重连风暴
+
 ---

 ## 配置说明
@@ -0,0 +1,92 @@
+package lion.Service;
+
+import io.netty.channel.Channel;
+import io.netty.util.AttributeKey;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 「哪条通道是主站」的唯一登记处。
+ *
+ * <p>历史缺陷：主站重连期间可能出现多条已认证通道，节点先引用 X，随后被后建立的 Y
+ * 覆盖；Y 断开时把引用清空，而仍可用的 X 继续发送心跳与任务消息，节点却因为引用为空
+ * 不再上报任何任务状态。主站侧的存活探测又能收到 X 上的响应，双方于是都判定
+ * 「连接正常」，未完成任务的状态就此永久卡住。
+ *
+ * <p>因此把引用管理收敛到这里，并保留全部已认证通道：首选通道一旦失效，立即回退到
+ * 其它仍可用的已认证通道；引用被误清时，任何一条已认证通道上的消息也能把它恢复。
+ * 这样「有已认证通道在，就一定能上报」成为不变式，不必等人工重连或下一个探测周期。
+ *
+ * <p>「已认证」的口径是「该通道发来过只有主站会发的消息类型」。事故中出问题的那条通道
+ * 正是靠可用性探测消息证明自己仍在工作的，若只认 {@code IdentityMessage}，这条通道
+ * 就永远无法重新取得引用。节点数据端口本就只应面向主站开放，凭证由网络边界提供。
+ */
+@Slf4j
+public class PrimaryChannelTracker {
+
+    static final AttributeKey<Boolean> AUTHENTICATED = AttributeKey.valueOf("primaryAuthenticated");
+
+    /** 全部已认证通道，用于在首选通道失效时立即回退。 */
+    private final Set<Channel> authenticated = ConcurrentHashMap.newKeySet();
+
+    private volatile Channel primary;
+
+    /** 收到只有主站会发的消息：认证该通道；若当前引用不是它，则改指向它。 */
+    public void onPrimaryMessage(Channel channel) {
+        markAuthenticated(channel);
+        if (primary == null) {
+            primary = channel;
+            log.info("主站通道引用已建立");
+        } else if (channel != primary) {
+            switchTo(channel);
+        }
+    }
+
+    private void markAuthenticated(Channel channel) {
+        if (Boolean.TRUE.equals(channel.attr(AUTHENTICATED).get()))
+            return;
+        channel.attr(AUTHENTICATED).set(Boolean.TRUE);
+        channel.closeFuture().addListener(ignored -> unregister(channel));
+        authenticated.add(channel);
+    }
+
+    /** 通道退出：移出已认证集合，必要时把引用移交给其它仍可用的通道。 */
+    public void unregister(Channel channel) {
+        authenticated.remove(channel);
+        if (channel != primary)
+            return;
+        primary = firstUsable();
+    }
+
+    private void switchTo(Channel channel) {
+        log.info("主站通道引用已切换到仍在工作的通道");
+        primary = channel;
+    }
+
+    public Channel current() {
+        Channel channel = primary;
+        if (channel != null && channel.isActive())
+            return channel;
+        // 首选通道已失效：立即回退，避免出现「有可用通道却无人上报」的静默状态。
+        Channel fallback = firstUsable();
+        if (fallback != null && fallback != channel) {
+            log.info("主站首选通道不可用，回退到其它已认证通道");
+            primary = fallback;
+        }
+        return fallback;
+    }
+
+    /** 引用是否存在且仍可写。 */
+    public boolean usable() {
+        return current() != null;
+    }
+
+    private Channel firstUsable() {
+        for (Channel channel : authenticated)
+            if (channel.isActive())
+                return channel;
+        return null;
+    }
+}
@@ -0,0 +1,34 @@
+package lion.Service;
+
+/**
+ * 「是否该主动重新唤起主站」的判定。
+ *
+ * <p>节点持有待上报的任务状态却没有可用通道时，只等主站下一个探测周期可能长达半小时；
+ * 期间用户在下载器里看到的是任务永远停在「已提交」。这里按固定间隔限流地允许重连，
+ * 既能在几十秒内自愈，又不会在主站确实离线时形成重连风暴。
+ *
+ * <p>把判定单独拆出来是为了能直接对时间轴写断言，不必启动 Netty。
+ */
+public class RekickPolicy {
+
+    /** 两次主动唤起之间的最小间隔。 */
+    static final long MIN_INTERVAL_MILLIS = 30_000L;
+
+    private long lastAttemptMillis;
+
+    /**
+     * 本轮是否应当唤起主站。
+     *
+     * @param pendingReports 是否有待上报的任务状态
+     * @param channelUsable  是否已存在可用通道
+     * @param nowMillis      当前时间
+     */
+    public boolean shouldRekick(boolean pendingReports, boolean channelUsable, long nowMillis) {
+        if (!pendingReports || channelUsable)
+            return false;
+        if (lastAttemptMillis != 0 && nowMillis - lastAttemptMillis < MIN_INTERVAL_MILLIS)
+            return false;
+        lastAttemptMillis = nowMillis;
+        return true;
+    }
+}
@@ -7,6 +7,8 @@ import lion.Message.Main.*;
 import lion.Service.DeleteService;
 import lion.Service.SubscriptionSnapshotStore;
 import lion.Service.DownloadCheckService;
+import lion.Service.PrimaryChannelTracker;
+import lion.Service.RekickPolicy;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.*;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -25,8 +27,23 @@ import java.util.concurrent.locks.ReentrantLock;
@Slf4j
 public class storageNode {

-    // 分别由 Netty IO 线程写入、由 5 秒定时线程读取，必须保证可见性。
-    volatile Channel server;
+    // 主站通道引用由 PrimaryChannelTracker 统一管理，避免重复认证把可用通道误清空。
+    final PrimaryChannelTracker primaryChannel = new PrimaryChannelTracker();
+
+    // 有待上报任务却无可用通道时，主动重新唤起主站，避免干等下一个探测周期。
+    final RekickPolicy rekickPolicy = new RekickPolicy();
+
+    // 重新唤起主站是阻塞式端口探测（最多 21×3 秒），不得占用 5 秒调度线程，
+    // 否则下载进度扫描与压缩都会被拖住；同一时刻只允许一次探测。
+    final java.util.concurrent.atomic.AtomicBoolean rekicking = new java.util.concurrent.atomic.AtomicBoolean();
+
+    // 专门用于阻塞式重连探测；不能复用单线程的 checkThreadPool，
+    // 在同一个线程池里提交阻塞任务会把调度线程自己堵住。
+    final ExecutorService rekickExecutor = Executors.newSingleThreadExecutor(r -> {
+        Thread thread = new Thread(r, "storage-node-rekick");
+        thread.setDaemon(true);
+        return thread;
+    });

    volatile Channel node;

@@ -72,6 +89,45 @@ public class storageNode {
                }).bind(real_port);
        log.info("listening on port {}", real_port);

+        wakeMainServer();
+        downloadCheckService = new DownloadCheckService(queue);
+        checkThreadPool = Executors.newScheduledThreadPool(1);
+        checkThreadPool.scheduleAtFixedRate(this::mainThread, 5, 5, TimeUnit.SECONDS);
+    }
+
+    /**
+     * 在独立线程里重新唤起主站。
+     *
+     * <p>{@link #wakeMainServer()} 是阻塞的端口探测，最坏可占 21×3 秒；直接在
+     * {@link #mainThread()} 里调用会让下载扫描与压缩停摆，因此投递到独立线程执行。
+     * 该执行器与 5 秒调度器分离，避免阻塞任务把调度线程自己堵住；
+     * 同一时刻只允许一次探测，防止主站长时间离线时堆积。
+     */
+    void rekickMainServerAsync(){
+        if (!rekicking.compareAndSet(false, true))
+            return;
+        try {
+            rekickExecutor.execute(() -> {
+                try {
+                    wakeMainServer();
+                } finally {
+                    rekicking.set(false);
+                }
+            });
+        } catch (Exception e) {
+            rekicking.set(false);
+            log.warn("安排主站重连失败:{}", e.getMessage());
+        }
+    }
+
+    /**
+     * 按既定顺序敲主站监听端口，请它回来建立数据通道。
+     *
+     * <p>启动时调用一次，另在「有待上报任务但无可用通道」时按 {@link RekickPolicy}
+     * 限流重试。主站侧的可用性探测只能覆盖它认为连接仍在的情况；一旦双方都判定
+     * 对方不在，只有本方法能重新建立通道，否则任务状态会一直卡在未完成。
+     */
+    void wakeMainServer(){
        int i;
        for(i=0; i<=20; i++) {
            try (Socket socket = new Socket()) {
@@ -88,9 +144,6 @@ public class storageNode {
        if (i==20) {
            log.info("server connect failed");
        }
-        downloadCheckService = new DownloadCheckService(queue);
-        checkThreadPool = Executors.newScheduledThreadPool(1);
-        checkThreadPool.scheduleAtFixedRate(this::mainThread, 5, 5, TimeUnit.SECONDS);
    }

    public void mainThread(){
@@ -104,6 +157,19 @@ public class storageNode {
            } finally {
                lock.unlock();
            }
+
+            // 有暂存内容却无可用通道：告警，并按限流主动重新唤起主站。
+            // 这里只告警与重连，不提前返回——本地下载与压缩必须继续推进，
+            // 否则主站恢复前任务会在节点侧整体停摆。
+            // 该判断在 downloadCheck 之前：正在下载、进度无变化的任务稍后会走
+            // 「无需上报」的提前返回路径，若只在回报分支里判定，通道失联后
+            // 这类任务既不会被发现也不会触发重连，就复现了历史上的静默卡死。
+            if (!queue.isEmpty() && !primaryChannel.usable()) {
+                log.warn("主站通道不可用，{} 个任务状态暂存待上报", queue.size());
+                if (rekickPolicy.shouldRekick(true, false, System.currentTimeMillis()))
+                    rekickMainServerAsync();
+            }
+
            //检查，当任务状态发生变化即方法返回true时，再更新，否则return
            if (!downloadCheckService.downloadCheck()) {
                boolean isSkip = true;
@@ -116,17 +182,16 @@ public class storageNode {
                        }
                if(isSkip) {
                    counter++;
-                    if (server != null && server.isActive() && counter > 10) {
-                        server.writeAndFlush(new MaintainMessage());
+                    if (primaryChannel.usable() && counter > 10) {
+                        primaryChannel.current().writeAndFlush(new MaintainMessage());
                        counter = 0;
                    }
                    return;
                }
            }
-            //主站未连接时不上报。原先直接 server.writeAndFlush 会在 server 为 null 时抛 NPE，
-            //被外层 catch 吞掉后每 5 秒刷一条 error；队列保留，等主站重连时重放。
-            Channel target = server;
-            if (target == null || !target.isActive())
+            //主站未连接时不上报，队列保留、等主站重连时重放（无法写通道时上面的分支已兜住）。
+            Channel target = primaryChannel.current();
+            if (target == null)
                return;
            //没有待上报任务时不必发送空数组（此时 downloadCheck 为 false 且无压缩完成任务）。
            if (queue.isEmpty())
@@ -151,6 +216,27 @@ public class storageNode {

    int counter;

+    /**
+     * 该消息是否只有主站会发。
+     *
+     * <p>白名单而非黑名单：只有这些类型能证明对端是主站并据此认领通道引用。
+     * 名单严格等于主站在 {@code RemoteService} 里实际会发出的类型；
+     * {@code ResponseMessage}、{@code DownloadStatusMessage}、{@code MaintainMessage}
+     * 都是本节点自己的出站类型，备机（{@code lionwebsiteside}）也会发身份消息，
+     * 都不能用来认领主站引用。
+     */
+    static boolean isPrimaryEvidence(AbstractMessage message) {
+        if (message.messageType == AbstractMessage.IDENTITY_MESSAGE) {
+            // 备机也用身份消息自我介绍，只有 identity 为 lionwebsite 的才是主站。
+            IdentityMessage identity = (IdentityMessage) message;
+            return "lionwebsite".equals(identity.getIdentity());
+        }
+        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;
+    }
+
    class MyChannelInboundHandlerAdapter extends ChannelInboundHandlerAdapter{
        Map<Integer, GalleryTask> queue;

@@ -166,17 +252,21 @@ public class storageNode {
                log.info(String.valueOf(msg));
            AbstractMessage abstractMessage = (AbstractMessage) msg;

-            // 来自已认证主站通道的任何消息都是「主站仍在」的证据。
-            // 备机的过期语义是「主站失联」，因此用这个信号刷新新鲜度，而不是只在收到快照时刷新：
-            // 订阅内容长期不变时主站没有理由重发整份快照，若只看快照就会误判过期。
-            if (ctx.channel().equals(server))
+            // 只有主站会发的消息类型（身份/任务下发/删除/探活/订阅快照）同时承担两件事：
+            // 证明「主站仍在」，以及恢复通道引用。并发认证时引用可能先指向 X 再被 Y 覆盖，
+            // Y 断开后引用被清空，而仍可用的 X 继续发送探活等消息；收到这些消息即证明
+            // X 就是主站，必须无条件认领——引用被清空时 current() 为 null，
+            // 任何以其为前提的判断都不成立，自愈也就无从触发。
+            if (isPrimaryEvidence(abstractMessage)) {
+                primaryChannel.onPrimaryMessage(ctx.channel());
                subscriptionSnapshotStore.markPrimaryContact();
+            }

            switch (abstractMessage.messageType){
                case AbstractMessage.IDENTITY_MESSAGE -> {
                    IdentityMessage identityMessage = (IdentityMessage) abstractMessage;
                    if(identityMessage.getIdentity().equals("lionwebsite")) {
-                        server = ctx.channel();
+                        // 引用已在上面按消息类型认领，这里只保留「上线」日志语义。
                        log.info("server 上线");
                    } else if(identityMessage.getIdentity().equals("lionwebsiteside")){
                        node = ctx.channel();
@@ -191,7 +281,7 @@ public class storageNode {
                        GalleryTask currentTask = downloadCheckService.addToQueue(dpm.getGalleryTask());
                        DownloadStatusMessage downloadStatusMessage = new DownloadStatusMessage();
                        downloadStatusMessage.setGalleryTasks(new GalleryTask[]{currentTask});
-                        Channel target = server;
+                        Channel target = primaryChannel.current();
                        if (target != null && target.isActive())
                            target.writeAndFlush(downloadStatusMessage);
                        log.info(String.valueOf(queue));
@@ -222,7 +312,7 @@ public class storageNode {
                }
                case AbstractMessage.SUBSCRIPTION_SNAPSHOT_MESSAGE -> {
                    SubscriptionSnapshotMessage snapshotMessage = (SubscriptionSnapshotMessage) abstractMessage;
-                    if (!Config.subscriptionSyncEnabled || !ctx.channel().equals(server)) {
+                    if (!Config.subscriptionSyncEnabled || !ctx.channel().equals(primaryChannel.current())) {
                        ctx.writeAndFlush(new ResponseMessage(snapshotMessage.messageId, SubscriptionSnapshotStore.APPLY_INVALID));
                        return;
                    }
@@ -241,13 +331,13 @@ public class storageNode {

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) {
-            if(ctx.channel().equals(server)) {
+            if(ctx.channel().equals(primaryChannel.current())) {
                log.info("server 下线");
-                server = null;
            } else if(ctx.channel().equals(node)){
                log.info("node 下线");
                node = null;
            }
+            primaryChannel.unregister(ctx.channel());
        }
    }

@@ -0,0 +1,88 @@
+package lion;
+
+import io.netty.channel.embedded.EmbeddedChannel;
+import lion.Message.AbstractMessage;
+import lion.Message.Main.AvailableCheckMessage;
+import lion.Message.Main.IdentityMessage;
+import lion.Service.PrimaryChannelTracker;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * 复现并锁定线上事故的通道恢复语义。
+ *
+ * <p>事故经过：主站重连期间节点先后认证了多条通道，引用被最后一条覆盖；那条通道断开后
+ * 引用被清空，而更早建立、仍然可用的通道继续发送可用性探测。节点因引用为空不再上报
+ * 任何任务状态，主站又能在旧通道上收到探活响应，双方都判定连接正常，未完成任务
+ * 便永久停在「已提交」，直到人工触发重连。
+ */
+class PrimaryChannelRecoveryTest {
+
+    @Test
+    void availableCheckOnSurvivingChannelRestoresReporting() {
+        PrimaryChannelTracker tracker = new PrimaryChannelTracker();
+        EmbeddedChannel survivor = new EmbeddedChannel();
+        EmbeddedChannel overwriter = new EmbeddedChannel();
+        try {
+            // 主站两次上线：引用最终落在后建立的 overwriter 上。
+            tracker.onPrimaryMessage(survivor);
+            tracker.onPrimaryMessage(overwriter);
+
+            // overwriter 断开，引用被清空——事故的起点。
+            tracker.unregister(overwriter);
+
+            // 仍然可用的 survivor 发来探活消息：引用应当立即恢复。
+            AvailableCheckMessage probe = new AvailableCheckMessage();
+            assertEquals(AbstractMessage.AVAILABLE_CHECK_MESSAGE, probe.messageType);
+            tracker.onPrimaryMessage(survivor);
+
+            assertSame(survivor, tracker.current(), "探活消息应恢复引用");
+            assertTrue(tracker.usable(), "恢复后必须能继续上报任务状态");
+        } finally {
+            survivor.finishAndReleaseAll();
+            overwriter.finishAndReleaseAll();
+        }
+    }
+
+    /** 首选通道断开但仍有其它已认证通道时，不得出现「无通道可用」的空窗。 */
+    @Test
+    void referenceNeverDropsWhileAnotherAuthenticatedChannelLives() {
+        PrimaryChannelTracker tracker = new PrimaryChannelTracker();
+        EmbeddedChannel first = new EmbeddedChannel();
+        EmbeddedChannel second = new EmbeddedChannel();
+        try {
+            tracker.onPrimaryMessage(first);
+            tracker.onPrimaryMessage(second);
+
+            tracker.unregister(second);
+
+            assertTrue(tracker.usable(), "仍有已认证通道时必须可用");
+            assertSame(first, tracker.current());
+        } finally {
+            first.finishAndReleaseAll();
+            second.finishAndReleaseAll();
+        }
+    }
+
+    /**
+     * 备机的身份消息不得夺走主站引用。
+     *
+     * <p>备机在启动时同样会自我介绍，若按消息类型无条件认领，备机通道会被误当成主站，
+     * 任务状态将上报到错误的连接上。
+     */
+    @Test
+    void sidecarIdentityDoesNotStealPrimaryReference() {
+        assertTrue(storageNode.isPrimaryEvidence(identity("lionwebsite")), "主站身份应被认作证据");
+        assertFalse(storageNode.isPrimaryEvidence(identity("lionwebsiteside")),
+                "备机身份不是主站证据，否则任务状态会上报到错误的连接");
+        assertFalse(storageNode.isPrimaryEvidence(new lion.Message.Main.ResponseMessage()),
+                "响应消息是节点自己的出站类型，不能用来认领主站引用");
+    }
+
+    private static IdentityMessage identity(String value) {
+        IdentityMessage message = new IdentityMessage();
+        message.setIdentity(value);
+        return message;
+    }
+}
@@ -0,0 +1,117 @@
+package lion.Service;
+
+import io.netty.channel.Channel;
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class PrimaryChannelTrackerTest {
+
+    /**
+     * 未认证通道退出时不得清空引用。
+     *
+     * <p>这正是线上事故的成因：并发认证让引用先指向 X 再被 Y 覆盖，
+     * Y 退出时把引用清空，而仍在工作的 X 从此不再被用于上报任务状态。
+     */
+    @Test
+    void unregisterOfNonPrimaryChannelKeepsReference() {
+        PrimaryChannelTracker tracker = new PrimaryChannelTracker();
+        EmbeddedChannel primary = new EmbeddedChannel();
+        EmbeddedChannel other = new EmbeddedChannel();
+        try {
+            tracker.onPrimaryMessage(primary);
+            tracker.onPrimaryMessage(other);
+            tracker.onPrimaryMessage(primary); // 引用回到仍在工作的那条通道
+
+            tracker.unregister(other);
+
+            assertSame(primary, tracker.current());
+            assertTrue(tracker.usable());
+        } finally {
+            primary.finishAndReleaseAll();
+            other.finishAndReleaseAll();
+        }
+    }
+
+    /**
+     * 首选通道退出时立即回退到其它仍可用的已认证通道。
+     *
+     * <p>事故中节点引用的那条通道退出后引用被清空，而另一条已认证通道仍然活跃，
+     * 任务状态就此无人上报。回退保证「有已认证通道在，就一定能上报」。
+     */
+    @Test
+    void unregisterOfPrimaryFallsBackToSurvivingChannel() {
+        PrimaryChannelTracker tracker = new PrimaryChannelTracker();
+        EmbeddedChannel survivor = new EmbeddedChannel();
+        EmbeddedChannel transientChannel = new EmbeddedChannel();
+        try {
+            tracker.onPrimaryMessage(survivor);
+            tracker.onPrimaryMessage(transientChannel);
+            tracker.unregister(transientChannel);
+
+            assertSame(survivor, tracker.current(), "应回退到仍可用的已认证通道");
+            assertTrue(tracker.usable());
+        } finally {
+            survivor.finishAndReleaseAll();
+            transientChannel.finishAndReleaseAll();
+        }
+    }
+
+    /** 实际在收发消息的已认证通道应当接管引用，即使原引用仍处于 active 状态。 */
+    @Test
+    void authenticatedChannelTakesOverOnMessage() {
+        PrimaryChannelTracker tracker = new PrimaryChannelTracker();
+        EmbeddedChannel stalePrimary = new EmbeddedChannel();
+        EmbeddedChannel live = new EmbeddedChannel();
+        try {
+            tracker.onPrimaryMessage(stalePrimary);
+            tracker.onPrimaryMessage(live);
+            tracker.onPrimaryMessage(stalePrimary); // 引用先落在 stalePrimary 上
+
+            tracker.onPrimaryMessage(live);
+
+            assertSame(live, tracker.current(), "说话的通道应成为引用");
+        } finally {
+            stalePrimary.finishAndReleaseAll();
+            live.finishAndReleaseAll();
+        }
+    }
+
+    /** 未经认证的通道不得占用引用，避免陌生连接抢走状态上报通道。 */
+    @Test
+    void unauthenticatedChannelCannotClaimReference() {
+        PrimaryChannelTracker tracker = new PrimaryChannelTracker();
+        EmbeddedChannel authenticated = new EmbeddedChannel();
+        EmbeddedChannel stranger = new EmbeddedChannel();
+        try {
+            tracker.onPrimaryMessage(authenticated);
+
+            // 未认证通道不会进入已认证集合，因此不能用它认领。
+
+            assertSame(authenticated, tracker.current());
+        } finally {
+            authenticated.finishAndReleaseAll();
+            stranger.finishAndReleaseAll();
+        }
+    }
+
+    /** 已下线的通道即使曾被认证，也不能再占用引用。 */
+    @Test
+    void inactiveChannelIsNotUsable() {
+        PrimaryChannelTracker tracker = new PrimaryChannelTracker();
+        EmbeddedChannel channel = new EmbeddedChannel();
+        tracker.onPrimaryMessage(channel);
+        channel.finishAndReleaseAll();
+
+        assertFalse(tracker.usable());
+    }
+
+    @Test
+    void emptyTrackerHasNoUsableChannel() {
+        PrimaryChannelTracker tracker = new PrimaryChannelTracker();
+
+        assertNull(tracker.current());
+        assertFalse(tracker.usable());
+    }
+}
@@ -0,0 +1,47 @@
+package lion.Service;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class RekickPolicyTest {
+
+    @Test
+    void noPendingReportsNeverRekicks() {
+        RekickPolicy policy = new RekickPolicy();
+
+        assertFalse(policy.shouldRekick(false, false, 1_000L));
+    }
+
+    @Test
+    void usableChannelNeverRekicks() {
+        RekickPolicy policy = new RekickPolicy();
+
+        assertFalse(policy.shouldRekick(true, true, 1_000L));
+    }
+
+    @Test
+    void pendingWithoutChannelRekicksThenRateLimits() {
+        RekickPolicy policy = new RekickPolicy();
+        long now = 1_000_000L;
+
+        assertTrue(policy.shouldRekick(true, false, now), "首次应允许重连");
+        assertFalse(policy.shouldRekick(true, false, now + 1), "间隔内应被限流");
+        assertFalse(policy.shouldRekick(true, false, now + RekickPolicy.MIN_INTERVAL_MILLIS - 1));
+        assertTrue(policy.shouldRekick(true, false, now + RekickPolicy.MIN_INTERVAL_MILLIS),
+                "超过最小间隔后应再次允许");
+    }
+
+    /** 限流只抑制连击，不应把「已限流」永久锁死后续尝试。 */
+    @Test
+    void rateLimitDoesNotBlockForever() {
+        RekickPolicy policy = new RekickPolicy();
+        long now = 500_000L;
+
+        policy.shouldRekick(true, false, now);
+        for (int i = 0; i < 5; i++)
+            policy.shouldRekick(true, false, now + i * 1_000L);
+
+        assertTrue(policy.shouldRekick(true, false, now + 10 * RekickPolicy.MIN_INTERVAL_MILLIS));
+    }
+}