修复订阅备机同步的一致性与重试问题

src/main/java/com/lion/lionwebsite/Message/SubscriptionSnapshotMessage.java

@@ -2,9 +2,11 @@ package com.lion.lionwebsite.Message;
 
 import lombok.Data;
 import lombok.NoArgsConstructor;
+import lombok.ToString;
 
 @Data
 @NoArgsConstructor
+@ToString(exclude = {"payloadBase64", "signature"})
 public class SubscriptionSnapshotMessage extends AbstractMessage {
     {
         messageType = SUBSCRIPTION_SNAPSHOT_MESSAGE;

src/main/java/com/lion/lionwebsite/Service/RemoteService.java

@@ -12,11 +12,13 @@ import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import io.netty.handler.logging.LoggingHandler;
+import io.netty.handler.logging.ByteBufFormat;
 import io.netty.util.concurrent.DefaultPromise;
 import io.netty.util.concurrent.Promise;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
 import jakarta.annotation.PostConstruct;
@@ -80,6 +82,8 @@ public class RemoteService {
 
     final AtomicBoolean subscriptionSyncQueued = new AtomicBoolean();
 
+    final AtomicBoolean subscriptionSyncRunning = new AtomicBoolean();
+
     @Value("${subscription.standby.sync-enabled:false}")
     boolean subscriptionSyncEnabled;
 
@@ -104,7 +108,8 @@ public class RemoteService {
                                 protected void initChannel(NioSocketChannel channel) {
                                     channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(100000000, 1, 4));
                                     channel.pipeline().addLast(new MessageCodec());
-                                    channel.pipeline().addLast(new LoggingHandler());
+                                    // 只记录事件和字节数，避免把订阅正文、签名等消息内容写入日志。
+                                    channel.pipeline().addLast(new LoggingHandler(io.netty.handler.logging.LogLevel.DEBUG, ByteBufFormat.SIMPLE));
                                     channel.pipeline().addLast(new MyChannelInboundHandlerAdapter());
                                 }
                             }).connect(new InetSocketAddress(ip, port + i)).sync();
@@ -176,35 +181,59 @@ public class RemoteService {
     public void requestSubscriptionSync() {
         if (!subscriptionSyncEnabled)
             return;
-        if (!subscriptionSyncQueued.compareAndSet(false, true))
+        subscriptionSyncQueued.set(true);
+        if (!subscriptionSyncRunning.compareAndSet(false, true))
             return;
-        subscriptionSyncExecutor.execute(() -> {
+        subscriptionSyncExecutor.execute(this::drainSubscriptionSyncQueue);
+    }
+
+    @Scheduled(fixedDelayString = "${subscription.standby.retry-interval-ms:60000}")
+    void scheduledSubscriptionSync() {
+        requestSubscriptionSync();
+    }
+
+    private void drainSubscriptionSyncQueue() {
         try {
-                do {
+            while (subscriptionSyncQueued.getAndSet(false)) {
-                    subscriptionSyncQueued.set(false);
                 if (isDead())
                     continue;
-                    SubscriptionSnapshotMessage message = subscriptionStandbySnapshotService.build();
+                syncSubscriptionSnapshotOnce();
+            }
+        } finally {
+            subscriptionSyncRunning.set(false);
+            if (subscriptionSyncQueued.get())
+                requestSubscriptionSync();
+        }
+    }
+
+    private void syncSubscriptionSnapshotOnce() {
+        SubscriptionSnapshotMessage message = null;
+        DefaultPromise<AbstractMessage> promise = null;
+        try {
+            message = subscriptionStandbySnapshotService.build();
             message.setMessageId(atomicInteger.getAndIncrement());
-                    DefaultPromise<AbstractMessage> promise = new DefaultPromise<>(eventLoopGroup);
+            promise = new DefaultPromise<>(eventLoopGroup);
             promiseHashMap.put(message.messageId, promise);
             channel.writeAndFlush(message);
             if (promise.await(30, TimeUnit.SECONDS)) {
                 AbstractMessage reply = promise.getNow();
-                        if (reply instanceof ResponseMessage response && response.getResult() == 0)
+                if (reply instanceof ResponseMessage response && (response.getResult() == 0 || response.getResult() == 3))
-                            log.info("订阅快照同步成功 revision={}", shortRevision(message.getRevision()));
+                    log.info("订阅快照同步完成 revision={} result={}", shortRevision(message.getRevision()), response.getResult());
                 else
                     log.warn("订阅快照同步失败 revision={} result={}", shortRevision(message.getRevision()),
-                                    reply instanceof ResponseMessage response ? response.getResult() : "timeout");
+                            reply instanceof ResponseMessage response ? response.getResult() : "invalid-response");
             } else {
                 log.warn("订阅快照同步超时 revision={}", shortRevision(message.getRevision()));
             }
-                    promiseHashMap.remove(message.messageId, promise);
+        } catch (InterruptedException e) {
-                } while (subscriptionSyncQueued.getAndSet(false));
+            Thread.currentThread().interrupt();
+            log.warn("订阅快照同步线程被中断");
         } catch (Exception e) {
             log.warn("生成或发送订阅快照失败: {}", e.getMessage());
+        } finally {
+            if (message != null && promise != null)
+                promiseHashMap.remove(message.messageId, promise);
         }
-        });
     }
 
     private static String shortRevision(String revision) {

src/main/java/com/lion/lionwebsite/Service/SubService.java

@@ -23,6 +23,8 @@ import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.Date;
+import java.util.concurrent.locks.Lock;
+import java.util.function.Supplier;
 
 @Service
 @Slf4j
@@ -32,8 +34,13 @@ public class SubService {
     final UserMapper userMapper;
     final SubscriptionRefreshService refreshService;
     final RemoteService remoteService;
+    final SubscriptionStateCoordinator stateCoordinator;
 
     public String insertSubscriptionAccount(String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
+        return withWriteLock(() -> insertSubscriptionAccountUnlocked(name, upstreamKey, filterHighMultiplier, enabled));
+    }
+
+    private String insertSubscriptionAccountUnlocked(String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
         Response response = Response.generateResponse();
         if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank())
             return response.failure("名称和上游 key 不能为空").toJSONString();
@@ -57,6 +64,10 @@ public class SubService {
     }
 
     public String updateSubscriptionAccount(Integer id, String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
+        return withWriteLock(() -> updateSubscriptionAccountUnlocked(id, name, upstreamKey, filterHighMultiplier, enabled));
+    }
+
+    private String updateSubscriptionAccountUnlocked(Integer id, String name, String upstreamKey, boolean filterHighMultiplier, boolean enabled) {
         Response response = Response.generateResponse();
         SubscriptionAccount account = subMapper.selectSubscriptionAccount(id);
         if (account == null)
@@ -79,6 +90,10 @@ public class SubService {
     }
 
     public String deleteSubscriptionAccount(Integer id) {
+        return withWriteLock(() -> deleteSubscriptionAccountUnlocked(id));
+    }
+
+    private String deleteSubscriptionAccountUnlocked(Integer id) {
         Response response = Response.generateResponse();
         SubscriptionAccount account = subMapper.selectSubscriptionAccount(id);
         if (account == null)
@@ -91,12 +106,20 @@ public class SubService {
     }
 
     public String refreshSubscriptionAccount(Integer id) {
+        return withWriteLock(() -> refreshSubscriptionAccountUnlocked(id));
+    }
+
+    private String refreshSubscriptionAccountUnlocked(Integer id) {
         boolean success = refreshService.refresh(id);
         remoteService.requestSubscriptionSync();
         return success ? Response._success("刷新成功") : Response._failure("刷新失败，请查看子账号错误状态");
     }
 
     public String insertSubBind(String user, Integer accountId) {
+        return withWriteLock(() -> insertSubBindUnlocked(user, accountId));
+    }
+
+    private String insertSubBindUnlocked(String user, Integer accountId) {
         Response response = Response.generateResponse();
         if (user == null || user.isBlank() || userMapper.selectUserByUsername(user) == null)
             return response.failure("用户不存在").toJSONString();
@@ -116,6 +139,10 @@ public class SubService {
     }
 
     public String resetKey(String user) {
+        return withWriteLock(() -> resetKeyUnlocked(user));
+    }
+
+    private String resetKeyUnlocked(String user) {
         Response response = Response.generateResponse();
         if (subMapper.countSubBindByUser(user) == 0)
             return response.failure("绑定不存在").toJSONString();
@@ -129,6 +156,10 @@ public class SubService {
     }
 
     public String rebind(String user, Integer accountId) {
+        return withWriteLock(() -> rebindUnlocked(user, accountId));
+    }
+
+    private String rebindUnlocked(String user, Integer accountId) {
         Response response = Response.generateResponse();
         SubscriptionAccount account = subMapper.selectSubscriptionAccount(accountId);
         if (account == null || !account.isEnabled())
@@ -193,12 +224,26 @@ public class SubService {
     }
 
     public String deleteSubBind(String user) {
+        return withWriteLock(() -> deleteSubBindUnlocked(user));
+    }
+
+    private String deleteSubBindUnlocked(String user) {
         subMapper.deleteSubBind(user);
         subMapper.deleteSubUpdateRecord(user);
         remoteService.requestSubscriptionSync();
         return Response._success("删除成功");
     }
 
+    private String withWriteLock(Supplier<String> action) {
+        Lock stateLock = stateCoordinator.writeLock();
+        stateLock.lock();
+        try {
+            return action.get();
+        } finally {
+            stateLock.unlock();
+        }
+    }
+
     private SubscriptionAccount firstEnabledAccount() {
         return subMapper.selectAllSubscriptionAccounts().stream().filter(SubscriptionAccount::isEnabled).findFirst().orElse(null);
     }

src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java

@@ -19,6 +19,7 @@ import java.nio.file.*;
 import java.util.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import java.util.concurrent.locks.Lock;
 
 @Service
 @Slf4j
@@ -29,6 +30,8 @@ public class SubscriptionRefreshService {
 
     final SubMapper subMapper;
 
+    final SubscriptionStateCoordinator stateCoordinator;
+
     @Value("${subscription.upstream.v2-url-template}")
     String v2UrlTemplate;
 
@@ -48,6 +51,9 @@ public class SubscriptionRefreshService {
     }
 
     public boolean refresh(Integer accountId) {
+        Lock stateLock = stateCoordinator.writeLock();
+        stateLock.lock();
+        try {
         SubscriptionAccount account = subMapper.selectSubscriptionAccount(accountId);
         if (account == null || !account.isEnabled())
             return false;
@@ -66,6 +72,9 @@ public class SubscriptionRefreshService {
             log.error("刷新子账号订阅失败 accountId={}: {}", accountId, message);
             return false;
         }
+        } finally {
+            stateLock.unlock();
+        }
     }
 
     public String v2Url(SubscriptionAccount account) {
@@ -204,11 +213,15 @@ public class SubscriptionRefreshService {
     }
 
     public void invalidateCache(Integer accountId) {
+        Lock stateLock = stateCoordinator.writeLock();
+        stateLock.lock();
         try {
             Files.deleteIfExists(cachedPath(accountId, "v2"));
             Files.deleteIfExists(cachedPath(accountId, "cat"));
         } catch (IOException e) {
             log.warn("清理失效订阅缓存失败 accountId={}", accountId, e);
+        } finally {
+            stateLock.unlock();
         }
     }
 }

src/main/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotService.java

@@ -9,6 +9,7 @@ import com.lion.lionwebsite.Util.CustomUtil;
 import lombok.RequiredArgsConstructor;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
+import jakarta.annotation.PostConstruct;
 
 import javax.crypto.Mac;
 import javax.crypto.spec.SecretKeySpec;
@@ -20,18 +21,32 @@ import java.nio.file.Path;
 import java.security.MessageDigest;
 import java.util.*;
 import java.util.zip.GZIPOutputStream;
+import java.util.concurrent.locks.Lock;
 
 @Service
 @RequiredArgsConstructor
 public class SubscriptionStandbySnapshotService {
     private final SubMapper subMapper;
     private final SubscriptionRefreshService refreshService;
+    private final SubscriptionStateCoordinator stateCoordinator;
     private final ObjectMapper objectMapper = CustomUtil.objectMapper;
 
     @Value("${subscription.standby.sync-secret:}")
     String syncSecret;
 
+    @Value("${subscription.standby.sync-enabled:false}")
+    boolean syncEnabled;
+
+    @PostConstruct
+    void validateConfiguration() {
+        if (syncEnabled && (syncSecret == null || syncSecret.isBlank()))
+            throw new IllegalStateException("启用订阅备机同步时必须配置 SUBSCRIPTION_SYNC_SECRET");
+    }
+
     public SubscriptionSnapshotMessage build() throws IOException {
+        Lock stateLock = stateCoordinator.readLock();
+        stateLock.lock();
+        try {
         Map<Integer, SubscriptionAccountSnapshot> accountMap = new HashMap<>();
         for (SubscriptionAccount account : subMapper.selectAllSubscriptionAccounts()) {
             if (!account.isEnabled() || !refreshService.hasCompleteCache(account.getId()))
@@ -83,6 +98,9 @@ public class SubscriptionStandbySnapshotService {
         message.setPayloadSha256(payloadSha256);
         message.setSignature(hmac(signatureInput.getBytes(StandardCharsets.UTF_8)));
         return message;
+        } finally {
+            stateLock.unlock();
+        }
     }
 
     private String hmac(byte[] input) {

src/main/java/com/lion/lionwebsite/Service/SubscriptionStateCoordinator.java

@@ -0,0 +1,20 @@
+package com.lion.lionwebsite.Service;
+
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/** Coordinates database bindings and the two cache files as one subscription state. */
+@Component
+public final class SubscriptionStateCoordinator {
+    private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
+
+    public Lock readLock() {
+        return lock.readLock();
+    }
+
+    public Lock writeLock() {
+        return lock.writeLock();
+    }
+}

src/main/resources/application.yaml

@@ -46,6 +46,7 @@ subscription:
   standby:
     sync-enabled: "${SUBSCRIPTION_STANDBY_SYNC_ENABLED:false}"
     sync-secret: "${SUBSCRIPTION_SYNC_SECRET:}"
+    retry-interval-ms: 60000
 
 bot:
   token: "5222939329:AAHa6l9ZuVVdNSDLPI_H-c8O_VgeOEw5plA"

src/test/java/com/lion/lionwebsite/Service/SubscriptionStandbySnapshotServiceTest.java

@@ -0,0 +1,63 @@
+package com.lion.lionwebsite.Service;
+
+import com.lion.lionwebsite.Dao.normal.SubMapper;
+import com.lion.lionwebsite.Domain.SubBind;
+import com.lion.lionwebsite.Domain.SubscriptionAccount;
+import com.lion.lionwebsite.Message.SubscriptionSnapshotMessage;
+import com.lion.lionwebsite.Message.SubscriptionSnapshotPayload;
+import com.lion.lionwebsite.Util.CustomUtil;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.zip.GZIPInputStream;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class SubscriptionStandbySnapshotServiceTest {
+    @Test
+    void buildsDeterministicSnapshotWithoutPlainPublicKeys(@TempDir Path directory) throws Exception {
+        SubMapper mapper = mock(SubMapper.class);
+        SubscriptionRefreshService refreshService = mock(SubscriptionRefreshService.class);
+        SubscriptionStateCoordinator coordinator = new SubscriptionStateCoordinator();
+        SubscriptionAccount account = new SubscriptionAccount(1, "account", "upstream-secret", true, true,
+                null, null, null, null, 2, null, null);
+        when(mapper.selectAllSubscriptionAccounts()).thenReturn(new ArrayList<>(java.util.List.of(account)));
+        when(mapper.selectAllSubBind()).thenReturn(new ArrayList<>(java.util.List.of(
+                new SubBind("public-key-a", "user-a", 1, "account", true, true),
+                new SubBind("public-key-b", "user-b", 1, "account", true, true))));
+        Path v2 = directory.resolve("v2ray.txt");
+        Path clash = directory.resolve("clash.yaml");
+        Files.writeString(v2, "v2-content", StandardCharsets.UTF_8);
+        Files.writeString(clash, "clash-content", StandardCharsets.UTF_8);
+        when(refreshService.hasCompleteCache(1)).thenReturn(true);
+        when(refreshService.cachedPath(1, "v2")).thenReturn(v2);
+        when(refreshService.cachedPath(1, "cat")).thenReturn(clash);
+
+        SubscriptionStandbySnapshotService service = new SubscriptionStandbySnapshotService(mapper, refreshService, coordinator);
+        service.syncSecret = "test-sync-secret";
+        SubscriptionSnapshotMessage first = service.build();
+        SubscriptionSnapshotMessage second = service.build();
+
+        assertEquals(first.getRevision(), second.getRevision());
+        byte[] compressed = Base64.getDecoder().decode(first.getPayloadBase64());
+        byte[] json;
+        try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(compressed))) {
+            json = gzip.readAllBytes();
+        }
+        String rawPayload = new String(json, StandardCharsets.UTF_8);
+        assertFalse(rawPayload.contains("public-key-a"));
+        assertFalse(rawPayload.contains("public-key-b"));
+        assertFalse(rawPayload.contains("upstream-secret"));
+        SubscriptionSnapshotPayload payload = CustomUtil.objectMapper.readValue(json, SubscriptionSnapshotPayload.class);
+        assertEquals(1, payload.getAccounts().size());
+        assertEquals(2, payload.getBindings().size());
+        assertTrue(payload.getAccounts().getFirst().isFilterHighMultiplier());
+    }
+}