diff --git a/src/main/java/com/lion/lionwebsite/Service/SubService.java b/src/main/java/com/lion/lionwebsite/Service/SubService.java index d779ce5..215ee9c 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubService.java +++ b/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("刷新失败，请查看子账号错误状态"); diff --git a/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java b/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java index 75f472e..1717075 100644 --- a/src/main/java/com/lion/lionwebsite/Service/SubscriptionRefreshService.java +++ b/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 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 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 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 download(String url) throws IOException { + List 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) { diff --git a/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshServiceTest.java b/src/test/java/com/lion/lionwebsite/Service/SubscriptionRefreshServiceTest.java new file mode 100644 index 0000000..34388ed --- /dev/null +++ b/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 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 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(); } + } +}