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