缩短订阅状态锁范围并拒绝过期刷新结果

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

@@ -37,17 +37,20 @@ public class SubService {
     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();
-        if (subMapper.countSubscriptionAccountName(name.trim()) > 0 || subMapper.countSubscriptionAccountKey(upstreamKey.trim()) > 0)
-            return response.failure("名称或上游 key 已存在").toJSONString();
-        SubscriptionAccount account = new SubscriptionAccount(null, name.trim(), upstreamKey.trim(), filterHighMultiplier, enabled, null, null, null, null, 0, null, null);
-        subMapper.insertSubscriptionAccount(account);
+        SubscriptionAccount account;
+        Lock lock = stateCoordinator.writeLock();
+        lock.lock();
+        try {
+            if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank())
+                return response.failure("名称和上游 key 不能为空").toJSONString();
+            if (subMapper.countSubscriptionAccountName(name.trim()) > 0 || subMapper.countSubscriptionAccountKey(upstreamKey.trim()) > 0)
+                return response.failure("名称或上游 key 已存在").toJSONString();
+            account = new SubscriptionAccount(null, name.trim(), upstreamKey.trim(), filterHighMultiplier, enabled, null, null, null, null, 0, null, null);
+            subMapper.insertSubscriptionAccount(account);
+        } finally {
+            lock.unlock();
+        }
         if (enabled)
             refreshService.refresh(account.getId());
         remoteService.requestSubscriptionSync();
@@ -64,27 +67,34 @@ 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)
-            return response.failure("子账号不存在").toJSONString();
-        if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank())
-            return response.failure("名称和上游 key 不能为空").toJSONString();
-        for (SubscriptionAccount existing : subMapper.selectAllSubscriptionAccounts()) {
-            if (!existing.getId().equals(id) && (existing.getName().equals(name.trim()) || existing.getUpstreamKey().equals(upstreamKey.trim())))
-                return response.failure("名称或上游 key 已存在").toJSONString();
+        SubscriptionAccount account;
+        Lock lock = stateCoordinator.writeLock();
+        lock.lock();
+        try {
+            account = subMapper.selectSubscriptionAccount(id);
+            if (account == null)
+                return response.failure("子账号不存在").toJSONString();
+            if (name == null || name.isBlank() || upstreamKey == null || upstreamKey.isBlank())
+                return response.failure("名称和上游 key 不能为空").toJSONString();
+            for (SubscriptionAccount existing : subMapper.selectAllSubscriptionAccounts()) {
+                if (!existing.getId().equals(id) && (existing.getName().equals(name.trim()) || existing.getUpstreamKey().equals(upstreamKey.trim())))
+                    return response.failure("名称或上游 key 已存在").toJSONString();
+            }
+            boolean changed = !account.getUpstreamKey().equals(upstreamKey.trim())
+                    || account.isFilterHighMultiplier() != filterHighMultiplier || account.isEnabled() != enabled;
+            account.setName(name.trim());
+            account.setUpstreamKey(upstreamKey.trim());
+            account.setFilterHighMultiplier(filterHighMultiplier);
+            account.setEnabled(enabled);
+            subMapper.updateSubscriptionAccount(account);
+            if (changed)
+                refreshService.invalidateCache(id);
+        } finally {
+            lock.unlock();
         }
-        account.setName(name.trim());
-        account.setUpstreamKey(upstreamKey.trim());
-        account.setFilterHighMultiplier(filterHighMultiplier);
-        account.setEnabled(enabled);
-        subMapper.updateSubscriptionAccount(account);
-        if (enabled && !refreshService.refresh(id))
-            refreshService.invalidateCache(id);
+        if (enabled)
+            refreshService.refresh(id);
         remoteService.requestSubscriptionSync();
         return response.success(accountJson(account)).toJSONString();
     }
@@ -101,15 +111,12 @@ public class SubService {
         if (account.getBoundUserCount() != null && account.getBoundUserCount() > 0)
             return response.failure("子账号仍绑定用户，请先改绑").toJSONString();
         subMapper.deleteSubscriptionAccount(id);
+        refreshService.invalidateCache(id);
         remoteService.requestSubscriptionSync();
         return response.success("删除成功").toJSONString();
     }
 
     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("刷新失败，请查看子账号错误状态");

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

@@ -5,6 +5,7 @@ import com.lion.lionwebsite.Domain.SubscriptionAccount;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.http.HttpEntity;
+import org.apache.http.client.config.RequestConfig;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.impl.client.CloseableHttpClient;
@@ -25,7 +26,10 @@ import java.util.concurrent.locks.Lock;
 @Slf4j
 @RequiredArgsConstructor
 public class SubscriptionRefreshService {
-    private static final CloseableHttpClient HTTP_CLIENT = HttpClients.createDefault();
+    private static final CloseableHttpClient HTTP_CLIENT = HttpClients.custom()
+            .setDefaultRequestConfig(RequestConfig.custom().setConnectTimeout(5_000)
+                    .setConnectionRequestTimeout(5_000).setSocketTimeout(15_000).build())
+            .build();
     private static final Pattern MULTIPLIER = Pattern.compile("(\\d+(?:\\.\\d+)?)x\\s*$", Pattern.CASE_INSENSITIVE);
 
     final SubMapper subMapper;
@@ -53,31 +57,63 @@ public class SubscriptionRefreshService {
         return success;
     }
 
+    private long refreshSequence;
+    // Accessed only while holding the coordinator write lock.
+    private final Map<Integer, Long> latestRefresh = new HashMap<>();
+
     public boolean refresh(Integer accountId) {
         Lock stateLock = stateCoordinator.writeLock();
+        SubscriptionAccount account;
+        long version;
         stateLock.lock();
         try {
-        SubscriptionAccount account = subMapper.selectSubscriptionAccount(accountId);
-        if (account == null || !account.isEnabled())
-            return false;
-        try {
-            String v2 = processV2(firstLine(download(v2Url(account))), account.isFilterHighMultiplier(), highMultiplierThreshold);
-            List<String> clash = processClash(download(clashUrl(account)), account.isFilterHighMultiplier(), highMultiplierThreshold);
-            Path dir = Paths.get(cacheRoot, String.valueOf(accountId));
-            Files.createDirectories(dir);
-            atomicWrite(dir.resolve("v2ray.txt"), v2.getBytes(StandardCharsets.UTF_8));
-            atomicWrite(dir.resolve("clash.yaml"), String.join("\n", clash).concat("\n").getBytes(StandardCharsets.UTF_8));
-            subMapper.markSubscriptionRefreshSuccess(accountId);
-            return true;
-        } catch (Exception e) {
-            String message = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
-            subMapper.markSubscriptionRefreshFailure(accountId, message.length() > 500 ? message.substring(0, 500) : message);
-            log.error("刷新子账号订阅失败 accountId={}: {}", accountId, message);
-            return false;
-        }
+            account = subMapper.selectSubscriptionAccount(accountId);
+            if (account == null || !account.isEnabled())
+                return false;
+            version = ++refreshSequence;
+            latestRefresh.put(accountId, version);
         } finally {
             stateLock.unlock();
         }
+
+        // Network access and parsing never hold the shared subscription lock.
+        try {
+            String v2 = processV2(firstLine(download(v2Url(account))), account.isFilterHighMultiplier(), highMultiplierThreshold);
+            List<String> clash = processClash(download(clashUrl(account)), account.isFilterHighMultiplier(), highMultiplierThreshold);
+            stateLock.lock();
+            try {
+                if (!isCurrent(account, version))
+                    return false;
+                Path dir = Paths.get(cacheRoot, String.valueOf(accountId));
+                Files.createDirectories(dir);
+                atomicWrite(dir.resolve("v2ray.txt"), v2.getBytes(StandardCharsets.UTF_8));
+                atomicWrite(dir.resolve("clash.yaml"), String.join("\n", clash).concat("\n").getBytes(StandardCharsets.UTF_8));
+                subMapper.markSubscriptionRefreshSuccess(accountId);
+                return true;
+            } finally {
+                stateLock.unlock();
+            }
+        } catch (Exception e) {
+            stateLock.lock();
+            try {
+                if (isCurrent(account, version)) {
+                    String message = e.getMessage() == null ? e.getClass().getSimpleName() : e.getMessage();
+                    subMapper.markSubscriptionRefreshFailure(accountId, message.length() > 500 ? message.substring(0, 500) : message);
+                }
+            } finally {
+                stateLock.unlock();
+            }
+            log.warn("刷新子账号订阅失败 accountId={} errorType={}", accountId, e.getClass().getSimpleName());
+            return false;
+        }
+    }
+
+    private boolean isCurrent(SubscriptionAccount expected, long version) {
+        SubscriptionAccount current = subMapper.selectSubscriptionAccount(expected.getId());
+        return Objects.equals(latestRefresh.get(expected.getId()), version)
+                && current != null && current.isEnabled()
+                && Objects.equals(current.getUpstreamKey(), expected.getUpstreamKey())
+                && current.isFilterHighMultiplier() == expected.isFilterHighMultiplier();
     }
 
     public String v2Url(SubscriptionAccount account) {
@@ -179,7 +215,7 @@ public class SubscriptionRefreshService {
         return matcher.find() && Double.parseDouble(matcher.group(1)) > threshold;
     }
 
-    private static List<String> download(String url) throws IOException {
+    List<String> download(String url) throws IOException {
         HttpGet get = new HttpGet(url);
         try (CloseableHttpResponse response = HTTP_CLIENT.execute(get)) {
             if (response.getStatusLine().getStatusCode() != 200)
@@ -219,6 +255,7 @@ public class SubscriptionRefreshService {
         Lock stateLock = stateCoordinator.writeLock();
         stateLock.lock();
         try {
+            latestRefresh.remove(accountId); // In-flight responses must not restore invalidated content.
             Files.deleteIfExists(cachedPath(accountId, "v2"));
             Files.deleteIfExists(cachedPath(accountId, "cat"));
         } catch (IOException e) {

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

@@ -0,0 +1,53 @@
+package com.lion.lionwebsite.Service;
+
+import com.lion.lionwebsite.Dao.normal.SubMapper;
+import com.lion.lionwebsite.Domain.SubscriptionAccount;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import java.nio.file.*;
+import java.util.List;
+import java.util.concurrent.*;
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+class SubscriptionRefreshServiceTest {
+    @Test
+    void stalledDownloadDoesNotHoldStateLockAndStaleResultIsDiscarded(@TempDir Path directory) throws Exception {
+        SubMapper mapper = mock(SubMapper.class);
+        SubscriptionAccount original = new SubscriptionAccount(1, "sample", "old", false, true,
+                null, null, null, null, 0, null, null);
+        SubscriptionAccount changed = new SubscriptionAccount(1, "sample", "new", false, true,
+                null, null, null, null, 0, null, null);
+        when(mapper.selectSubscriptionAccount(1)).thenReturn(original);
+        var coordinator = new SubscriptionStateCoordinator();
+        CountDownLatch started = new CountDownLatch(1);
+        CountDownLatch release = new CountDownLatch(1);
+        var service = new SubscriptionRefreshService(mapper, coordinator) {
+            @Override List<String> download(String url) throws java.io.IOException {
+                started.countDown();
+                try {
+                    if (!release.await(5, TimeUnit.SECONDS)) throw new java.io.IOException("test timed out");
+                } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new java.io.IOException(e); }
+                return url.contains("v2") ? List.of("bm9kZQ==") : List.of("proxies:");
+            }
+        };
+        service.v2UrlTemplate = "https://example.invalid/v2/{key}";
+        service.clashUrlTemplate = "https://example.invalid/clash/{key}";
+        service.cacheRoot = directory.toString();
+        ExecutorService worker = Executors.newSingleThreadExecutor();
+        try {
+            Future<Boolean> refresh = worker.submit(() -> service.refresh(1));
+            assertTrue(started.await(2, TimeUnit.SECONDS));
+            var lock = coordinator.writeLock();
+            assertTrue(lock.tryLock(1, TimeUnit.SECONDS), "management must remain available during downloads");
+            try {
+                when(mapper.selectSubscriptionAccount(1)).thenReturn(changed);
+                service.invalidateCache(1);
+            } finally { lock.unlock(); }
+            release.countDown();
+            assertFalse(refresh.get(2, TimeUnit.SECONDS));
+            assertFalse(service.hasCompleteCache(1));
+            verify(mapper, never()).markSubscriptionRefreshSuccess(any());
+        } finally { release.countDown(); worker.shutdownNow(); }
+    }
+}