From 9faae09851f79cc6c58a169dcf2d0de0a338bee4 Mon Sep 17 00:00:00 2001 From: root Date: Wed, 7 Oct 2026 23:50:52 +0800 Subject: [PATCH] fix: reconcile v2 retries and parse all gallery search layouts --- .../lionwebsite/Service/RemoteService.java | 71 +++++-- .../lionwebsite/Util/GallerySearchParser.java | 43 +++- .../Service/RemoteServiceEligibilityTest.java | 195 +++++++++++++++++- .../GallerySearchParserRegressionTest.java | 58 ++++++ 4 files changed, 342 insertions(+), 25 deletions(-) create mode 100644 src/test/java/com/lion/lionwebsite/Util/GallerySearchParserRegressionTest.java diff --git a/src/main/java/com/lion/lionwebsite/Service/RemoteService.java b/src/main/java/com/lion/lionwebsite/Service/RemoteService.java index 975cf17..15f63c8 100644 --- a/src/main/java/com/lion/lionwebsite/Service/RemoteService.java +++ b/src/main/java/com/lion/lionwebsite/Service/RemoteService.java @@ -614,6 +614,13 @@ public class RemoteService { byte submitResult = addGalleryToQueue(gallery); if(submitResult != 0 && !statusFuture.isDone()) return new RetryResult(false, "节点未接受重试请求"); + // 竞态兜底：本方法读到的 Gallery 是调用方传入的旧快照，注册等待者与节点被通知之间 + // 节点可能刚完成清理（例如 ACK 之后已清队列、重复 DOWNLOAD 只回放 receipt 不再产生新上报）。 + // 此时按「当前库里的真实状态」收尾：只认重新读到的完成状态，绝不把传入的旧进行中状态 + // 当作节点的新确认返回。 + String completedStatus = currentCompletedStatus(gallery.getGid()); + if (completedStatus != null && !statusFuture.isDone()) + statusFuture.complete(completedStatus); return new RetryResult(true, statusFuture.get(10, TimeUnit.SECONDS)); }catch (TimeoutException e){ return new RetryResult(false, "节点已收到重试请求，但未及时返回任务状态"); @@ -638,6 +645,22 @@ public class RemoteService { waiters.forEach(waiter -> waiter.complete(status)); } + /** + * 重新读库确认任务是否已真正完成；只用于「节点已接受重试」之后的竞态兜底。 + * + *

读库失败一律返回 null：宁可继续等上报，也不能凭一次失败的读把未知状态说成完成。 + */ + private String currentCompletedStatus(int gid) { + try { + Gallery current = galleryMapper.selectGalleryByGid(gid); + if (current != null && TaskStatus.COMPLETE.equals(current.getStatus())) + return current.getStatus(); + } catch (RuntimeException e) { + log.warn("重试时读取任务当前状态失败 gid={}", gid, e); + } + return null; + } + public record RetryResult(boolean success, String message) {} public byte deleteGallery(Gallery gallery){ @@ -738,12 +761,7 @@ public class RemoteService { } } - /** - * 消费协议 v2 上报：短事务提交 → 事务成功后回 ACK → 再用独立执行器做通知/推送。 - * - *

顺序很关键：数据库失败时既不能回 ACK，也不能清任务；慢通知不能阻塞这里， - * 否则会拖住后续上报的提交。 - */ + /** 消费协议 v2 上报：事务提交 → 唤醒重试等待者 → 回 ACK → 独立执行器做通知/推送。 */ private void consumeV2Report(TaskReportMessage report) { try { TaskReportHandler handler = taskReportHandler; @@ -757,6 +775,8 @@ public class RemoteService { log.warn("v2 上报未提交，等待节点重投 reportId={}", report.getReportId()); return; } + // 先完成等待者：apply 已确认提交，这一步绝不能因随后的 ACK 写入抛异常而丢失。 + completeRetryStatusWaitersFor(committed.acceptedTasks()); // 提交成功之后才回执，让节点清理快照。 Channel target = channel; if (target != null && target.isActive()) @@ -767,6 +787,20 @@ public class RemoteService { } } + /** 事务提交之后、ACK 与通知之前，按「真正被接受的更新」唤醒等待重试结果的调用方。 + * 被过滤的过期/旧 operation/待删除任务不在 acceptedTasks 里，不得完成等待者。 */ + private void completeRetryStatusWaitersFor(List acceptedTasks) { + if (acceptedTasks == null || acceptedTasks.isEmpty()) + return; + for (GalleryTask task : acceptedTasks) { + if (task == null) + continue; + String status = TaskStatus.fromReport(task.getStatus()); + if (status != null) + completeRetryStatusWaiters(task.getGid(), status); + } + } + /** 通知/推送在独立执行器里执行，外部延迟不会阻塞状态提交。 */ private void dispatchNotifications(TaskReportMessage report, TaskReportHandler.CommittedReport committed) { // 只推送事务里真正被接受的更新（含下载中/压缩中等中间状态）； @@ -775,16 +809,21 @@ public class RemoteService { ? new GalleryTask[0] : committed.acceptedTasks().toArray(new GalleryTask[0]); List newlyCompleted = committed.newlyCompleted(); - notificationExecutor.execute(() -> { - try { - for (Gallery gallery : newlyCompleted) - pushService.downloadComplete(gallery); - if (accepted.length > 0) - webSocketService.updateTaskProcessing(accepted); - } catch (Exception e) { - log.warn("任务完成通知/推送失败", e); - } - }); + try { + notificationExecutor.execute(() -> { + try { + for (Gallery gallery : newlyCompleted) + pushService.downloadComplete(gallery); + if (accepted.length > 0) + webSocketService.updateTaskProcessing(accepted); + } catch (Exception e) { + log.warn("任务完成通知/推送失败", e); + } + }); + } catch (java.util.concurrent.RejectedExecutionException e) { + // 通知队列已关闭只是「推不出去」，既不能影响已提交的 ACK，也不能影响重试等待者。 + log.warn("通知任务被拒绝，状态仍已提交 reportId={}", report.getReportId()); + } } /** 把节点上报的任务状态写回库里并推送给前端，按上报顺序逐个处理。 */ diff --git a/src/main/java/com/lion/lionwebsite/Util/GallerySearchParser.java b/src/main/java/com/lion/lionwebsite/Util/GallerySearchParser.java index 656cb25..5a26e2d 100644 --- a/src/main/java/com/lion/lionwebsite/Util/GallerySearchParser.java +++ b/src/main/java/com/lion/lionwebsite/Util/GallerySearchParser.java @@ -8,12 +8,13 @@ import java.util.ArrayList; import java.util.List; import java.util.regex.Pattern; -/** Search rows are located by gallery links, independent of table nesting and headings. */ +/** Search entries support table and thumbnail layouts; metadata excludes title/tag text. */ public final class GallerySearchParser { private GallerySearchParser() { } public static List parse(Document document) throws IOException { List result = new ArrayList<>(); - for (Element row : document.select("tr")) { + for (Element row : document.select("tr, .gl1t")) { + if (row.tagName().equals("tr") && !row.select(".gl1t").isEmpty()) continue; Element link = row.selectFirst("a[href*=/g/]"); if (link == null) continue; String href = link.attr("href"); @@ -21,16 +22,25 @@ public final class GallerySearchParser { if (gid == null) continue; GalleryForQuery gallery = new GalleryForQuery(); gallery.setGid(gid.toString()); gallery.setLink(href); - gallery.setName(link.select(".glink").isEmpty() ? link.text() : link.select(".glink").text()); + Element title = row.selectFirst(".glink"); + String name = title == null ? "" : title.text(); + if (name.isBlank()) { + for (Element candidate : row.select("a[href*=/g/]")) { + if (!candidate.text().isBlank()) { + name = candidate.text(); + break; + } + } + } + gallery.setName(name); Element type = row.selectFirst(".cn, .cs"); - gallery.setType(type == null ? row.select("td").first().text() : type.text()); + Element firstCell = row.selectFirst("td"); + gallery.setType(type != null ? type.text() : firstCell == null ? "" : firstCell.text()); Element image = row.selectFirst("img"); String thumbnail = image == null ? "" : image.attr("src"); if (thumbnail.isBlank() || thumbnail.startsWith("data:")) thumbnail = image == null ? "" : image.attr("data-src"); gallery.setThumbnailUrl(thumbnail.replace("https://s.exhentai.org", "")); - var pages = Pattern.compile("(\\d+)\\s+pages?", Pattern.CASE_INSENSITIVE).matcher(row.text()); - if (!pages.find()) throw new IOException("上游搜索结果缺少页数 gid=" + gid); - gallery.setPage(Integer.parseInt(pages.group(1))); + gallery.setPage(pageCount(row, gid)); Element date = row.selectFirst("[id^=posted_]"); if (date != null) gallery.setUploadTime(date.text()); else { @@ -42,4 +52,23 @@ public final class GallerySearchParser { } return result; } + + private static int pageCount(Element row, int gid) throws IOException { + Element metadata = row.clone(); + // Extended rows can put title and metadata in the same cell: remove only + // the title links/tags, retaining their containing cells and page counts. + metadata.select("a[href*=/g/], .glink, .gt, .gtl, .gtd, .glthumb").remove(); + var expression = Pattern.compile("^(\\d+)\\s+pages?$", Pattern.CASE_INSENSITIVE); + for (Element element : metadata.getAllElements()) { + var pages = expression.matcher(element.text().strip()); + if (!pages.matches()) continue; + try { + int count = Integer.parseInt(pages.group(1)); + if (count > 0) return count; + } catch (NumberFormatException invalid) { + throw new IOException("上游搜索结果页数格式不支持 gid=" + gid, invalid); + } + } + throw new IOException("上游搜索结果缺少页数 gid=" + gid); + } } diff --git a/src/test/java/com/lion/lionwebsite/Service/RemoteServiceEligibilityTest.java b/src/test/java/com/lion/lionwebsite/Service/RemoteServiceEligibilityTest.java index 4d12419..bc04165 100644 --- a/src/test/java/com/lion/lionwebsite/Service/RemoteServiceEligibilityTest.java +++ b/src/test/java/com/lion/lionwebsite/Service/RemoteServiceEligibilityTest.java @@ -2,13 +2,21 @@ package com.lion.lionwebsite.Service; import com.lion.lionwebsite.Dao.normal.GalleryMapper; import com.lion.lionwebsite.Domain.Gallery; +import com.lion.lionwebsite.Domain.GalleryTask; import com.lion.lionwebsite.Message.ProtocolHelloMessage; import com.lion.lionwebsite.Message.TaskCommandMessage; +import com.lion.lionwebsite.Message.TaskReportAckMessage; +import com.lion.lionwebsite.Message.TaskReportMessage; import io.netty.channel.embedded.EmbeddedChannel; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.CopyOnWriteArrayList; @@ -26,20 +34,28 @@ import static org.mockito.Mockito.*; class RemoteServiceEligibilityTest { private GalleryMapper galleryMapper; + private PushService pushService; + private WebSocketService webSocketService; private RemoteService service; private EmbeddedChannel channel; + private ExecutorService retryExecutor; private final CopyOnWriteArrayList written = new CopyOnWriteArrayList<>(); private final java.util.concurrent.CountDownLatch helloWritten = new java.util.concurrent.CountDownLatch(1); + private volatile boolean failAckWrites; @BeforeEach void setUp() { galleryMapper = mock(GalleryMapper.class); - service = new RemoteService(galleryMapper, mock(PushService.class), - mock(WebSocketService.class), mock(SubscriptionStandbySnapshotService.class)); + pushService = mock(PushService.class); + webSocketService = mock(WebSocketService.class); + service = new RemoteService(galleryMapper, pushService, + webSocketService, mock(SubscriptionStandbySnapshotService.class)); channel = new EmbeddedChannel(new io.netty.channel.ChannelOutboundHandlerAdapter() { @Override public void write(io.netty.channel.ChannelHandlerContext ctx, Object msg, io.netty.channel.ChannelPromise promise) { + if (failAckWrites && msg instanceof TaskReportAckMessage) + throw new io.netty.channel.ChannelException("ACK 写通道失败"); io.netty.util.concurrent.Promise inner = service.promiseHashMap.get(((com.lion.lionwebsite.Message.AbstractMessage) msg).messageId); if (inner != null) @@ -54,11 +70,13 @@ class RemoteServiceEligibilityTest { io.netty.channel.ChannelFuture channelFuture = mock(io.netty.channel.ChannelFuture.class); when(channelFuture.channel()).thenReturn(channel); service.channelFuture = channelFuture; + retryExecutor = Executors.newCachedThreadPool(); } @AfterEach void tearDown() { service.shutdownResources(); + retryExecutor.shutdownNow(); channel.finishAndReleaseAll(); } @@ -70,6 +88,52 @@ class RemoteServiceEligibilityTest { return gallery; } + private static GalleryTask task(int gid, byte status, int proceeding) { + GalleryTask task = new GalleryTask(); + task.setGid(gid); + task.setName("任务 [" + gid + "]"); + task.setStatus(status); + task.setProceeding(proceeding); + return task; + } + + private static TaskReportMessage report(int gid, String reportId) { + TaskReportMessage report = new TaskReportMessage(); + report.setReportId(reportId); + report.setRevision(7); + report.setOperationIds(Map.of(gid, "op-" + gid)); + return report; + } + + /** 让节点命令立刻拿到成功的 ResponseMessage（沿用既有 setup 的自动应答行为）。 */ + private void enableReadyDispatch() { + service.setDispatchResolver(gid -> GalleryTaskLifecycleService.DispatchDecision.ready("op-" + gid)); + service.v2Negotiated = true; + } + + private Future retryInBackground(int gid) { + return retryExecutor.submit(() -> service.retryGallery(gallery(gid))); + } + + private void awaitWaiterRegistered(int gid) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(3); + while (System.nanoTime() < deadline) { + var waiters = service.retryStatusWaiters.get(gid); + if (waiters != null && !waiters.isEmpty()) + return; + Thread.sleep(5); + } + fail("重试等待者未注册 gid=" + gid); + } + + private List ackedReportIds() { + List ids = new java.util.ArrayList<>(); + for (Object message : written) + if (message instanceof TaskReportAckMessage ack) + ids.add(ack.getReportId()); + return ids; + } + /** 未协商 v2 / 装配未就绪时绝不批量重投未完成任务。 */ @Test void resetUndoneSkipsBeforeNegotiationAndWiring() { @@ -156,4 +220,131 @@ class RemoteServiceEligibilityTest { channel.writeInbound(new ProtocolHelloMessage()); assertTrue(negotiation.get(2, TimeUnit.SECONDS)); } + + /** 正常 v2 重试：节点上报提交成功后，等待者必须在短 deadline 内被唤醒，而不是拖到 10 秒超时。 */ + @Test + void v2RetryCompletesWithinShortDeadlineThroughRealHandler() throws Exception { + enableReadyDispatch(); + service.setTaskReportHandler(report -> new TaskReportHandler.CommittedReport( + report.getReportId(), List.of(), List.of(task(501, GalleryTask.COMPRESS_COMPLETE, 40)))); + + Future pending = retryInBackground(501); + awaitWaiterRegistered(501); + + TaskReportMessage message = report(501, "rep-501"); + message.setGalleryTasks(new GalleryTask[]{task(501, GalleryTask.COMPRESS_COMPLETE, 40)}); + long start = System.nanoTime(); + channel.writeInbound(message); + + RemoteService.RetryResult result = pending.get(3, TimeUnit.SECONDS); + long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + assertTrue(result.success(), result.message()); + assertEquals(TaskStatus.COMPLETE, result.message()); + assertTrue(elapsedMs < 2_000, "正常 v2 上报不应退化为 10 秒超时，实测 " + elapsedMs + "ms"); + } + + /** 未提交（apply 返回 null）不回执也不完成；空了 acceptedTasks 的任务同样不能完成等待者。 */ + @Test + void uncommittedOrFilteredReportNeverCompletesWaiter() throws Exception { + enableReadyDispatch(); + + service.setTaskReportHandler(rejectedHandler -> null); + Future uncommitted = retryInBackground(502); + awaitWaiterRegistered(502); + TaskReportMessage rejected = report(502, "rep-502"); + rejected.setGalleryTasks(new GalleryTask[]{task(502, GalleryTask.COMPRESS_COMPLETE, 40)}); + channel.writeInbound(rejected); + service.awaitStatusApplied(); + assertFalse(uncommitted.isDone(), "未提交的上报不能完成重试等待者"); + assertTrue(ackedReportIds().isEmpty(), "未提交的上报不能回执"); + + // 已被事务过滤掉（过期/旧 operation/待删除）：只 ACK，不完成等待者。 + service.setTaskReportHandler(filteredHandler -> new TaskReportHandler.CommittedReport( + filteredHandler.getReportId(), List.of(), List.of())); + Future filtered = retryInBackground(503); + awaitWaiterRegistered(503); + TaskReportMessage filteredMessage = report(503, "rep-503"); + filteredMessage.setGalleryTasks(new GalleryTask[]{task(503, GalleryTask.COMPRESS_COMPLETE, 40)}); + channel.writeInbound(filteredMessage); + service.awaitStatusApplied(); + assertFalse(filtered.isDone(), "被过滤掉的更新不能完成重试等待者"); + assertEquals(List.of("rep-503"), ackedReportIds(), "已提交的上报仍应回执，让节点清理快照"); + } + + /** 通知线程异常或通知队列拒绝都不得影响 ACK 与等待者完成。 */ + @Test + void notificationFailureDoesNotAffectAckOrWaiterCompletion() throws Exception { + enableReadyDispatch(); + doThrow(new RuntimeException("推送失败")).when(webSocketService) + .updateTaskProcessing(any(GalleryTask[].class)); + service.setTaskReportHandler(report -> new TaskReportHandler.CommittedReport( + report.getReportId(), List.of(), List.of(report.getGalleryTasks()))); + + Future pending = retryInBackground(504); + awaitWaiterRegistered(504); + TaskReportMessage message = report(504, "rep-504"); + message.setGalleryTasks(new GalleryTask[]{task(504, GalleryTask.COMPRESS_COMPLETE, 40)}); + channel.writeInbound(message); + + RemoteService.RetryResult result = pending.get(3, TimeUnit.SECONDS); + assertTrue(result.success(), result.message()); + assertEquals(TaskStatus.COMPLETE, result.message()); + assertTrue(ackedReportIds().contains("rep-504")); + + // 通知队列已拒绝：提交成功仍必须回执并完成等待者。 + service.notificationExecutor.shutdownNow(); + Future afterShutdown = retryInBackground(505); + awaitWaiterRegistered(505); + TaskReportMessage queued = report(505, "rep-505"); + queued.setGalleryTasks(new GalleryTask[]{task(505, GalleryTask.COMPRESS_COMPLETE, 40)}); + channel.writeInbound(queued); + service.awaitStatusApplied(); + + RemoteService.RetryResult queuedResult = afterShutdown.get(3, TimeUnit.SECONDS); + assertTrue(queuedResult.success(), queuedResult.message()); + assertTrue(ackedReportIds().contains("rep-505")); + } + + /** + * 节点 accepted 后刚清完队列、当前 DB 已是下载完成：必须用重新读到的完成状态收尾， + * 不能拿调用方传入的旧 Gallery（旧「下载中」）当成新确认。 + */ + @Test + void retryShortCircuitsOnFreshlyCompletedDbState() throws Exception { + enableReadyDispatch(); + service.setTaskReportHandler(report -> null); + Gallery completed = gallery(506); + completed.setStatus(TaskStatus.COMPLETE); + when(galleryMapper.selectGalleryByGid(506)).thenReturn(completed); + + Gallery stale = gallery(506); + stale.setStatus(TaskStatus.DOWNLOADING); + long start = System.nanoTime(); + RemoteService.RetryResult result = service.retryGallery(stale); + long elapsedMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); + + assertTrue(result.success(), result.message()); + assertEquals(TaskStatus.COMPLETE, result.message(), "必须返回重新读到的完成状态"); + assertTrue(elapsedMs < 2_000, "已完成的库状态应立刻返回，实测 " + elapsedMs + "ms"); + verify(galleryMapper).selectGalleryByGid(506); + } + + /** ACK 写通道抛异常也不得吞掉已提交的重试状态：等待者仍须拿到真实状态，而非 10 秒超时。 */ + @Test + void ackWriteFailureStillCompletesRetryWaiter() throws Exception { + enableReadyDispatch(); + service.setTaskReportHandler(report -> new TaskReportHandler.CommittedReport( + report.getReportId(), List.of(), List.of(report.getGalleryTasks()))); + failAckWrites = true; + + Future pending = retryInBackground(507); + awaitWaiterRegistered(507); + TaskReportMessage message = report(507, "rep-507"); + message.setGalleryTasks(new GalleryTask[]{task(507, GalleryTask.COMPRESS_COMPLETE, 40)}); + channel.writeInbound(message); + + RemoteService.RetryResult result = pending.get(3, TimeUnit.SECONDS); + assertTrue(result.success(), result.message()); + assertEquals(TaskStatus.COMPLETE, result.message()); + } } diff --git a/src/test/java/com/lion/lionwebsite/Util/GallerySearchParserRegressionTest.java b/src/test/java/com/lion/lionwebsite/Util/GallerySearchParserRegressionTest.java new file mode 100644 index 0000000..8950d59 --- /dev/null +++ b/src/test/java/com/lion/lionwebsite/Util/GallerySearchParserRegressionTest.java @@ -0,0 +1,58 @@ +package com.lion.lionwebsite.Util; + +import org.jsoup.Jsoup; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.*; + +class GallerySearchParserRegressionTest { + @Test + void thumbnailModeUsesGalleryCards() throws Exception { + var document = Jsoup.parse("
" + + "" + + "
" + + "
Title [100 pages]
" + + "
Manga
42 pages
" + + "
2026-10-07 12:00
"); + var galleries = GallerySearchParser.parse(document); + assertEquals(1, galleries.size()); + assertEquals("Title [100 pages]", galleries.get(0).getName()); + assertEquals(42, galleries.get(0).getPage()); + assertEquals("Manga", galleries.get(0).getType()); + assertEquals("/cover.jpg", galleries.get(0).getThumbnailUrl()); + } + + @Test + void extendedModeTakesTitleFromNameRatherThanCoverLink() throws Exception { + var document = Jsoup.parse("
" + + "" + + "
" + + "
Title
" + + "
Manga
42 pages
"); + var galleries = GallerySearchParser.parse(document); + assertEquals(1, galleries.size()); + assertEquals("Title", galleries.get(0).getName()); + assertEquals(42, galleries.get(0).getPage()); + } + + @Test + void titleAndTagsCannotOverridePageCount() throws Exception { + var document = Jsoup.parse("" + + "" + + "
Manga" + + "" + + "
Title [100 pages]
" + + "
200 pages
42 pages
"); + var galleries = GallerySearchParser.parse(document); + assertEquals(1, galleries.size()); + assertEquals(42, galleries.get(0).getPage()); + } + + @Test + void pageCountRequiresMetadataEvenIfTitleContainsPages() { + var document = Jsoup.parse("
Manga" + + "" + + "
Title [100 pages]
"); + assertThrows(java.io.IOException.class, () -> GallerySearchParser.parse(document)); + } +}