fix: reconcile v2 retries and parse all gallery search layouts

2026-10-07 23:50:52 +08:00
parent 4c1a2d22e2
commit 9faae09851
4 changed files with 342 additions and 25 deletions
@@ -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));
    }

+    /**
+     * 重新读库确认任务是否已真正完成；只用于「节点已接受重试」之后的竞态兜底。
+     *
+     * <p>读库失败一律返回 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 → 再用独立执行器做通知/推送。
-         *
-         * <p>顺序很关键：数据库失败时既不能回 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<GalleryTask> 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,6 +809,7 @@ public class RemoteService {
                    ? new GalleryTask[0]
                    : committed.acceptedTasks().toArray(new GalleryTask[0]);
            List<Gallery> newlyCompleted = committed.newlyCompleted();
+            try {
                notificationExecutor.execute(() -> {
                    try {
                        for (Gallery gallery : newlyCompleted)
@@ -785,6 +820,10 @@ public class RemoteService {
                        log.warn("任务完成通知/推送失败", e);
                    }
                });
+            } catch (java.util.concurrent.RejectedExecutionException e) {
+                // 通知队列已关闭只是「推不出去」，既不能影响已提交的 ACK，也不能影响重试等待者。
+                log.warn("通知任务被拒绝，状态仍已提交 reportId={}", report.getReportId());
+            }
        }

        /** 把节点上报的任务状态写回库里并推送给前端，按上报顺序逐个处理。 */
@@ -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<GalleryForQuery> parse(Document document) throws IOException {
        List<GalleryForQuery> 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);
+    }
 }
@@ -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<Object> 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<com.lion.lionwebsite.Message.AbstractMessage> 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<RemoteService.RetryResult> 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<String> ackedReportIds() {
+        List<String> 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<RemoteService.RetryResult> 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<RemoteService.RetryResult> 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<RemoteService.RetryResult> 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<RemoteService.RetryResult> 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<RemoteService.RetryResult> 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<RemoteService.RetryResult> 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());
+    }
 }
@@ -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("<div class=itg><div class=gl1t>"
+                + "<a href='https://exhentai.org/g/12/key/'><img src='/cover.jpg'></a>"
+                + "<div class=gl4t><a href='https://exhentai.org/g/12/key/'>"
+                + "<div class=glink>Title [100 pages]</div></a></div>"
+                + "<div class=cn>Manga</div><div class=gl5t>42 pages</div>"
+                + "<div id=posted_12>2026-10-07 12:00</div></div></div>");
+        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("<table><tr><td class=gl1e><div>"
+                + "<a href='https://exhentai.org/g/12/key/'><img src='/cover.jpg'></a>"
+                + "</div></td><td class='gl2e glname'>"
+                + "<a href='https://exhentai.org/g/12/key/'><div class=glink>Title</div></a>"
+                + "<div class=cn>Manga</div><div>42 pages</div></td></tr></table>");
+        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("<table><tr><td class=cn>Manga</td>"
+                + "<td><img src='/c.jpg'></td><td class=glname>"
+                + "<a href='https://exhentai.org/g/12/key/'>"
+                + "<div class=glink>Title [100 pages]</div></a>"
+                + "<div class=gt title='misc:200 pages'>200 pages</div></td>"
+                + "<td class=gl4c><div>42 pages</div></td></tr></table>");
+        var galleries = GallerySearchParser.parse(document);
+        assertEquals(1, galleries.size());
+        assertEquals(42, galleries.get(0).getPage());
+    }
+
+    @Test
+    void pageCountRequiresMetadataEvenIfTitleContainsPages() {
+        var document = Jsoup.parse("<table><tr><td class=cn>Manga</td><td class=glname>"
+                + "<a href='https://exhentai.org/g/12/key/'>"
+                + "<div class=glink>Title [100 pages]</div></a></td></tr></table>");
+        assertThrows(java.io.IOException.class, () -> GallerySearchParser.parse(document));
+    }
+}