历史缺陷：主站重连期间可能出现多条已认证通道，节点先引用 X，随后被后建立的 Y + * 覆盖；Y 断开时把引用清空，而仍可用的 X 继续发送心跳与任务消息，节点却因为引用为空 + * 不再上报任何任务状态。主站侧的存活探测又能收到 X 上的响应，双方于是都判定 + * 「连接正常」，未完成任务的状态就此永久卡住。 + * + *
因此把引用管理收敛到这里，并保留全部已认证通道：首选通道一旦失效，立即回退到 + * 其它仍可用的已认证通道；引用被误清时，任何一条已认证通道上的消息也能把它恢复。 + * 这样「有已认证通道在，就一定能上报」成为不变式，不必等人工重连或下一个探测周期。 + * + *
「已认证」的口径是「该通道发来过只有主站会发的消息类型」。事故中出问题的那条通道
+ * 正是靠可用性探测消息证明自己仍在工作的，若只认 {@code IdentityMessage}，这条通道
+ * 就永远无法重新取得引用。节点数据端口本就只应面向主站开放，凭证由网络边界提供。
+ */
+@Slf4j
+public class PrimaryChannelTracker {
+
+ static final AttributeKey 节点持有待上报的任务状态却没有可用通道时，只等主站下一个探测周期可能长达半小时；
+ * 期间用户在下载器里看到的是任务永远停在「已提交」。这里按固定间隔限流地允许重连，
+ * 既能在几十秒内自愈，又不会在主站确实离线时形成重连风暴。
+ *
+ * 把判定单独拆出来是为了能直接对时间轴写断言，不必启动 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;
+ }
+}
diff --git a/src/main/java/lion/storageNode.java b/src/main/java/lion/storageNode.java
index e710629..90ac98b 100644
--- a/src/main/java/lion/storageNode.java
+++ b/src/main/java/lion/storageNode.java
@@ -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);
+ }
+
+ /**
+ * 在独立线程里重新唤起主站。
+ *
+ * {@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());
+ }
+ }
+
+ /**
+ * 按既定顺序敲主站监听端口，请它回来建立数据通道。
+ *
+ * 启动时调用一次，另在「有待上报任务但无可用通道」时按 {@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;
+ /**
+ * 该消息是否只有主站会发。
+ *
+ * 白名单而非黑名单：只有这些类型能证明对端是主站并据此认领通道引用。
+ * 名单严格等于主站在 {@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 事故经过：主站重连期间节点先后认证了多条通道，引用被最后一条覆盖；那条通道断开后
+ * 引用被清空，而更早建立、仍然可用的通道继续发送可用性探测。节点因引用为空不再上报
+ * 任何任务状态，主站又能在旧通道上收到探活响应，双方都判定连接正常，未完成任务
+ * 便永久停在「已提交」，直到人工触发重连。
+ */
+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();
+ }
+ }
+
+ /**
+ * 备机的身份消息不得夺走主站引用。
+ *
+ * 备机在启动时同样会自我介绍，若按消息类型无条件认领，备机通道会被误当成主站，
+ * 任务状态将上报到错误的连接上。
+ */
+ @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;
+ }
+}
diff --git a/src/test/java/lion/Service/PrimaryChannelTrackerTest.java b/src/test/java/lion/Service/PrimaryChannelTrackerTest.java
new file mode 100644
index 0000000..ccb3dde
--- /dev/null
+++ b/src/test/java/lion/Service/PrimaryChannelTrackerTest.java
@@ -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 {
+
+ /**
+ * 未认证通道退出时不得清空引用。
+ *
+ * 这正是线上事故的成因：并发认证让引用先指向 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();
+ }
+ }
+
+ /**
+ * 首选通道退出时立即回退到其它仍可用的已认证通道。
+ *
+ * 事故中节点引用的那条通道退出后引用被清空，而另一条已认证通道仍然活跃，
+ * 任务状态就此无人上报。回退保证「有已认证通道在，就一定能上报」。
+ */
+ @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());
+ }
+}
diff --git a/src/test/java/lion/Service/RekickPolicyTest.java b/src/test/java/lion/Service/RekickPolicyTest.java
new file mode 100644
index 0000000..56e5dea
--- /dev/null
+++ b/src/test/java/lion/Service/RekickPolicyTest.java
@@ -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));
+ }
+}