diff --git a/src/main/java/lion/Service/SubscriptionSnapshotStore.java b/src/main/java/lion/Service/SubscriptionSnapshotStore.java index b84c813..2baf2fc 100644 --- a/src/main/java/lion/Service/SubscriptionSnapshotStore.java +++ b/src/main/java/lion/Service/SubscriptionSnapshotStore.java @@ -39,17 +39,20 @@ public final class SubscriptionSnapshotStore { private final ObjectMapper objectMapper = CustomUtil.objectMapper; private final AtomicReference current = new AtomicReference<>(); /** - * 最近一次收到并校验通过的主站快照的时刻(本地接收时间)。 + * 最近一次收到主站存活信号的时刻(本地接收时间)。 * - *

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

「过期」的语义是「主站已失联」,因此新鲜度必须由「主站的任何存活信号」驱动, + * 而不能只看「收到快照」。快照是内容寻址的:内容长期不变时主站没有理由反复推送, + * 若以收到快照为准,备机会在内容不动的第 7 天误判过期;反过来,若主站为此定时重发 + * 整份快照,又只为续期而传输约 346 KiB/次。 * - *

接收时间取自节点本地时钟、不参与签名,因此攻击者无法通过重放旧 revision 续期。 + *

主站本就有常规存活探测(每 30 分钟的 AvailableCheckMessage),节点回它以 + * ResponseMessage。把该信号纳入新鲜度后:过期 ⇔ 主站失联超过有效期, + * 语义精确,且不再需要任何为「续期」而生的专用推送。 + * + *

时刻取自节点本地时钟、不参与签名,攻击者无法通过重放旧 revision 续期。 */ - private volatile long lastSyncAt; + private volatile long lastPrimaryContactAt; public SubscriptionSnapshotStore(Path root, String syncSecret, long maxStaleSeconds, int maxPayloadBytes) { this.root = Objects.requireNonNull(root); @@ -86,7 +89,7 @@ public final class SubscriptionSnapshotStore { current.set(snapshot); // 重启后还没有收到过主站快照,先以落盘内容的生成时间作为新鲜度起点, // 否则重启即视为「刚同步过」,会让超期快照被错误续期。 - lastSyncAt = snapshot.generatedAt(); + lastPrimaryContactAt = snapshot.generatedAt(); writePointer(revision); log.info("加载订阅快照成功 revision={} accounts={} bindings={}", shortRevision(revision), snapshot.accountCount(), snapshot.bindingCount()); return; @@ -123,7 +126,7 @@ public final class SubscriptionSnapshotStore { if (old != null) { if (message.getRevision().equals(old.revision())) { // 内容未变但主站仍在同步:刷新新鲜度,避免备机因「内容长期不变」而误判过期。 - lastSyncAt = System.currentTimeMillis(); + lastPrimaryContactAt = System.currentTimeMillis(); return new ApplyResult(APPLY_OLD, "revision 已存在"); } if (message.getGeneratedAt() < old.generatedAt()) @@ -157,7 +160,7 @@ public final class SubscriptionSnapshotStore { Snapshot snapshot = new Snapshot(message.getRevision(), message.getGeneratedAt(), data.byKeyHash, data.accounts.size(), data.bindingCount); current.set(snapshot); - lastSyncAt = System.currentTimeMillis(); + lastPrimaryContactAt = System.currentTimeMillis(); cleanupOldSnapshots(message.getRevision()); return new ApplyResult(APPLY_SUCCESS, "同步成功"); } catch (Exception e) { @@ -194,13 +197,28 @@ public final class SubscriptionSnapshotStore { if (snapshot == null) return new Status("unavailable", null, 0, 0, 0); long now = System.currentTimeMillis(); - long age = Math.max(0, now - lastSyncAt); + long age = Math.max(0, now - lastPrimaryContactAt); return new Status(isStale(now) ? "expired" : "ready", snapshot.revision(), snapshot.accountCount(), snapshot.bindingCount(), age); } - /** 距最近一次成功接收主站快照是否已超过最大有效期。 */ + /** + * 主站是否已失联超过最大有效期。 + * + *

由主站的任何存活信号刷新(见 {@link #markPrimaryContact()}),不限于快照。 + */ private boolean isStale(long now) { - return maxStaleMillis > 0 && now - lastSyncAt > maxStaleMillis; + return maxStaleMillis > 0 && now - lastPrimaryContactAt > maxStaleMillis; + } + + /** + * 记录一次「主站仍在」的证据。 + * + *

用于快照之外的常规存活信号(例如主站每 30 分钟的可用性检查)。 + * 只要主站在线,本节点的新鲜度就会持续被刷新,因此备机不会因为 + * 「订阅内容长期不变、主站没理由重发快照」而被判过期。 + */ + public void markPrimaryContact() { + lastPrimaryContactAt = System.currentTimeMillis(); } private Snapshot loadSnapshot(Path directory) throws IOException { diff --git a/src/main/java/lion/storageNode.java b/src/main/java/lion/storageNode.java index d812adf..e710629 100644 --- a/src/main/java/lion/storageNode.java +++ b/src/main/java/lion/storageNode.java @@ -166,6 +166,12 @@ public class storageNode { log.info(String.valueOf(msg)); AbstractMessage abstractMessage = (AbstractMessage) msg; + // 来自已认证主站通道的任何消息都是「主站仍在」的证据。 + // 备机的过期语义是「主站失联」,因此用这个信号刷新新鲜度,而不是只在收到快照时刷新: + // 订阅内容长期不变时主站没有理由重发整份快照,若只看快照就会误判过期。 + if (ctx.channel().equals(server)) + subscriptionSnapshotStore.markPrimaryContact(); + switch (abstractMessage.messageType){ case AbstractMessage.IDENTITY_MESSAGE -> { IdentityMessage identityMessage = (IdentityMessage) abstractMessage; diff --git a/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java b/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java index 1e515c0..381e495 100644 --- a/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java +++ b/src/test/java/lion/Service/SubscriptionSnapshotStoreTest.java @@ -98,6 +98,43 @@ class SubscriptionSnapshotStoreTest { assertNull(reloaded.lookup("v2", "public-key-1"), "过期快照不得继续分发"); } + /** + * 主站的常规存活信号(例如每 30 分钟的可用性检查)必须刷新新鲜度。 + * + *

这是「过期 ⇔ 主站失联」的关键:快照是内容寻址的,内容长期不变时主站没有理由 + * 重发整份快照;若新鲜度只由「收到快照」驱动,备机会在内容不动的第 7 天误判过期。 + */ + @Test + void primaryContactWithoutNewSnapshotKeepsSubscriptionFresh(@TempDir Path directory) throws Exception { + SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 1, 1024 * 1024); + SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis()); + assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code()); + + // 超过有效期(1 秒),期间只有主站存活信号、没有新快照 + Thread.sleep(1_200); + assertEquals("expired", store.status().state(), "有效期已过且无任何主站信号,应先判过期"); + + store.markPrimaryContact(); + assertEquals("ready", store.status().state(), "收到主站存活信号后应恢复可用"); + assertNotNull(store.lookup("v2", "public-key-1"), "主站在线期间必须能取到订阅"); + // 内容仍是最初那份,说明续期靠的是存活信号,而不是内容变化 + assertArrayEquals("v2-content".getBytes(StandardCharsets.UTF_8), store.lookup("v2", "public-key-1").content()); + } + + /** 主站失联(长时间没有任何信号)后仍必须判过期,不能因加入存活信号而永不失效。 */ + @Test + void staysExpiredWhenPrimaryContactStops(@TempDir Path directory) throws Exception { + SubscriptionSnapshotStore store = new SubscriptionSnapshotStore(directory, SECRET, 1, 1024 * 1024); + SubscriptionSnapshotMessage message = message("public-key-1", "v2-content", "clash-content", System.currentTimeMillis()); + assertEquals(SubscriptionSnapshotStore.APPLY_SUCCESS, store.apply(message).code()); + store.markPrimaryContact(); + assertEquals("ready", store.status().state()); + + Thread.sleep(1_200); // 之后主站再无任何消息 + assertEquals("expired", store.status().state(), "主站失联超过有效期必须判过期"); + assertNull(store.lookup("v2", "public-key-1"), "失联后不得继续分发"); + } + private SubscriptionSnapshotMessage message(String publicKey, String v2, String clash, long generatedAt) throws Exception { byte[] v2Bytes = v2.getBytes(StandardCharsets.UTF_8); byte[] clashBytes = clash.getBytes(StandardCharsets.UTF_8);