diff --git a/src/main/java/lion/Service/DownloadCheckService.java b/src/main/java/lion/Service/DownloadCheckService.java index a06a2b4..248bd24 100644 --- a/src/main/java/lion/Service/DownloadCheckService.java +++ b/src/main/java/lion/Service/DownloadCheckService.java @@ -54,6 +54,28 @@ public class DownloadCheckService { private record VerifiedArchive(long size, long verifiedAt) {} + /** + * 状态调度使用的共享监视器就是本实例对象本身。 + * + *

删除命令与下载扫描争用同一份任务队列与磁盘目录,两者必须真正互斥:只检查 + * 「是否正在压缩」的状态判定不足以构成互斥。扫描线程可能在「标 COMPRESSING」与 + * 「进压缩队列」之间被抢占,而删除此刻看到的队列与 active 集合都还没有该 gid,于是删除 + * 先把画廊删掉,扫描随后把删除前选中的旧对象排进压缩队列,压缩线程又把归档重新写出来—— + * 已删除的画廊被「复活」并上报给主站。 + * + *

因此 {@link #downloadCheck()}、{@link #addToQueue}、{@link #removeGid}、 + * {@link #isCompressing} 的状态调度,以及 {@link #compress()} 对队列的「领取」,全部在 + * 本监视器下完成;{@code ReliableTaskService.executeDelete} 也用同一监视器把 + * 「查忙 → 删磁盘 → 清队列」包成一个原子段。 + * + *

锁定顺序(唯一方向,避免死锁):调用方锁 → 本实例监视器 → {@code compress_queue}。 + * 真正的压缩 I/O 不在监视器内执行,删除的互斥由 {@link #activeCompressions} 标记保证。 + * + *

共享监视器就是 {@code this}({@code DownloadCheckService} 实例)本身,因此 + * {@code ReliableTaskService.executeDelete} 用 {@code synchronized (downloadCheckService)} + * 即可与本类的状态调度取到同一把锁。 + */ + /** * 完成通知的专用线程。 * @@ -90,65 +112,72 @@ public class DownloadCheckService { } public boolean downloadCheck(){ - if(queue.isEmpty()) - return false; - log.info("下载检查:{}", Arrays.toString(queue.values().toArray())); + synchronized (this) { + if(queue.isEmpty()) + return false; + log.info("下载检查:{}", Arrays.toString(queue.values().toArray())); - File downloadDirectory = new File(downloadPath); - File[] fileArray = downloadDirectory.listFiles(); + File downloadDirectory = new File(downloadPath); + File[] fileArray = downloadDirectory.listFiles(); - if(fileArray == null || fileArray.length == 0) - return false; + if(fileArray == null || fileArray.length == 0) + return false; - ArrayList files = new ArrayList<>(Arrays.asList(fileArray)); + ArrayList files = new ArrayList<>(Arrays.asList(fileArray)); - boolean result = false; - //扫描进度 - Iterator fileIterator = files.iterator(); - for(GalleryTask galleryTask: queue.values()){ - //跳过已经下载完成或者压缩完成的任务 - if(galleryTask.is_compress_complete() || galleryTask.is_compressing()) { - result = true; - continue; - } - - while(fileIterator.hasNext()){ - File file = fileIterator.next(); - if(!file.isDirectory() || !matchesGid(file.getName(), galleryTask.getGid())) - continue; - - galleryTask.setStatus(GalleryTask.DOWNLOADING); - fileIterator.remove(); - if(galleryTask.getName() == null || !galleryTask.getName().equals(file.getName())) - galleryTask.setName(file.getName()); - - File[] pages = file.listFiles((dir, name) -> !name.equals("galleryinfo.txt")); - if (pages == null || pages.length == 0) - continue; - if(galleryTask.getProceeding() != pages.length) + boolean result = false; + //扫描进度 + Iterator fileIterator = files.iterator(); + for(GalleryTask galleryTask: queue.values()){ + //跳过已经下载完成或者压缩完成的任务 + if(galleryTask.is_compress_complete() || galleryTask.is_compressing()) { result = true; - galleryTask.setProceeding(pages.length); - - if (new File(file.getPath(), "galleryinfo.txt").exists()) { - result = true; - galleryTask.setStatus(GalleryTask.DOWNLOAD_COMPLETE); - galleryTask.setPath(file.getPath()); + continue; } + + while(fileIterator.hasNext()){ + File file = fileIterator.next(); + if(!file.isDirectory() || !matchesGid(file.getName(), galleryTask.getGid())) + continue; + + galleryTask.setStatus(GalleryTask.DOWNLOADING); + fileIterator.remove(); + if(galleryTask.getName() == null || !galleryTask.getName().equals(file.getName())) + galleryTask.setName(file.getName()); + + File[] pages = file.listFiles((dir, name) -> !name.equals("galleryinfo.txt")); + if (pages == null || pages.length == 0) + continue; + if(galleryTask.getProceeding() != pages.length) + result = true; + galleryTask.setProceeding(pages.length); + + if (new File(file.getPath(), "galleryinfo.txt").exists()) { + result = true; + galleryTask.setStatus(GalleryTask.DOWNLOAD_COMPLETE); + galleryTask.setPath(file.getPath()); + } + } + fileIterator = files.iterator(); } - fileIterator = files.iterator(); + + //压缩队列 + for(GalleryTask galleryTask: queue.values()) + if (galleryTask.is_download_complete() + && System.currentTimeMillis() >= retryAfter.getOrDefault(galleryTask.getGid(), 0L)) { + // 必须以队列里的当前对象为准:本轮遍历的引用可能已被删除流程 removeGid + // 移除(删除已把磁盘清掉),把旧对象排进压缩队列会让已删除的画廊复活。 + GalleryTask current = queue.get(galleryTask.getGid()); + if (current != galleryTask) + continue; + galleryTask.setStatus(GalleryTask.COMPRESSING); + synchronized (compress_queue) { + compress_queue.add(galleryTask); + } + } + + return result; } - - //压缩队列 - for(GalleryTask galleryTask: queue.values()) - if (galleryTask.is_download_complete() - && System.currentTimeMillis() >= retryAfter.getOrDefault(galleryTask.getGid(), 0L)) { - galleryTask.setStatus(GalleryTask.COMPRESSING); - synchronized (compress_queue) { - compress_queue.add(galleryTask); - } - } - - return result; } /** @@ -156,7 +185,9 @@ public class DownloadCheckService { */ public void compress() { ArrayList galleryTasks; - synchronized (compress_queue) { + // 领取与扫描入队共用同一监视器:原子地把队列交给压缩线程并标记 active, + // 使删除再也看不到「已标 COMPRESSING 但尚未 active」的空档。 + synchronized (this) { if (compress_queue.isEmpty()) return; galleryTasks = new ArrayList<>(compress_queue); @@ -210,14 +241,14 @@ public class DownloadCheckService { *

删除与压缩是互斥的:命中时调用方应回可重试的「忙」,而不是继续动磁盘。 */ public boolean isCompressing(int gid){ - if (activeCompressions.contains(gid)) - return true; - synchronized (compress_queue) { + synchronized (this) { + if (activeCompressions.contains(gid)) + return true; for (GalleryTask queued : compress_queue) if (queued.getGid() == gid) return true; + return false; } - return false; } /** @@ -226,9 +257,9 @@ public class DownloadCheckService { *

用于删除成功后清理本节点内部状态,避免已删除任务继续被扫描或压缩。 */ public void removeGid(int gid){ - queue.remove(gid); - retryAfter.remove(gid); - synchronized (compress_queue) { + synchronized (this) { + queue.remove(gid); + retryAfter.remove(gid); compress_queue.removeIf(task -> task.getGid() == gid); } } @@ -275,40 +306,69 @@ public class DownloadCheckService { } } + /** + * 受理或刷新一个任务,并按 gid 用磁盘事实重建它的状态。 + * + *

与扫描、删除共用同一监视器:这里既要读磁盘目录,又要写任务队列,如果与删除交叉执行, + * 会出现「删除已清空磁盘,入队仍按删除前的目录刷新出下载完成」的矛盾状态。 + * + *

重建顺序是「归档优先」:gid 稳定而目录名不稳定(下载器会追加后缀),因此先按 gid + * 找到校验有效的 ZIP 判为压缩完成,再回退到下载目录判断下载中/下载完成。只有归档有效才算 + * 终态,损坏或空归档一律不得被当成完成。 + */ public GalleryTask addToQueue(GalleryTask galleryTask){ - GalleryTask active = queue.get(galleryTask.getGid()); - if (active != null && active.is_compressing()) - return active; - retryAfter.remove(galleryTask.getGid()); - // A reconnect can resend a task whose name is stale (for example, the - // downloader appended a resolution suffix). Resolve completed archives - // by gid first, because the gid is stable while the directory name is not. - File storedDirectory = findStoredDirectoryByGid(galleryTask.getGid()); - if(storedDirectory != null){ - queue.remove(galleryTask.getGid()); - galleryTask.setName(storedDirectory.getName()); - galleryTask.setStatus(GalleryTask.COMPRESS_COMPLETE); - notifyAlreadyStored(galleryTask.getName()); - return galleryTask; - } + return addToQueue(galleryTask, true); + } - // If the task is still downloading, retain the actual directory name so - // subsequent progress and compression use the same identity. - File downloadingDirectory = findDirectoryByGid(new File(downloadPath), galleryTask.getGid()); - if(downloadingDirectory != null){ - GalleryTask queuedTask = queue.get(galleryTask.getGid()); - if(queuedTask == null){ - refreshDownloadingTask(galleryTask, downloadingDirectory); - queue.put(galleryTask.getGid(), galleryTask); - return galleryTask; + /** + * 受理或刷新一个任务的内部实现。 + * + *

{@code notifyOnAlreadyStored=false} 供启动恢复使用:恢复时对每个任务做一次磁盘事实 + * 重建,命中的是「重启前就已经存在的归档」,而不是主站刚刚新下的任务,不该为它再发一遍 + * 「添加时已下载完成」通知。 + */ + GalleryTask addToQueue(GalleryTask galleryTask, boolean notifyOnAlreadyStored){ + synchronized (this) { + GalleryTask active = queue.get(galleryTask.getGid()); + // 只有真正「正在压缩」才短路:压缩线程正持有该对象写归档,替换它会丢状态。 + if (active != null && active.is_compressing()) + return active; + retryAfter.remove(galleryTask.getGid()); + // A reconnect can resend a task whose name is stale (for example, the + // downloader appended a resolution suffix). Resolve completed archives + // by gid first, because the gid is stable while the directory name is not. + File storedDirectory = findStoredDirectoryByGid(galleryTask.getGid()); + if(storedDirectory != null){ + // 已有有效归档即为终态;保留队列里那个仍然活跃的对象,避免同一 gid 出现两个实例。 + GalleryTask target = active == null ? galleryTask : active; + if (active != null) + queue.remove(galleryTask.getGid()); + target.setName(storedDirectory.getName()); + target.setPath(storedDirectory.getPath()); + target.setStatus(GalleryTask.COMPRESS_COMPLETE); + if (notifyOnAlreadyStored) + notifyAlreadyStored(target.getName()); + return target; } - if(!queuedTask.is_compressing() && !queuedTask.is_compress_complete()) - refreshDownloadingTask(queuedTask, downloadingDirectory); - return queuedTask; - } - GalleryTask queuedTask = queue.putIfAbsent(galleryTask.getGid(), galleryTask); - return queuedTask == null ? galleryTask : queuedTask; + // If the task is still downloading, retain the actual directory name so + // subsequent progress and compression use the same identity. + File downloadingDirectory = findDirectoryByGid(new File(downloadPath), galleryTask.getGid()); + if(downloadingDirectory != null){ + GalleryTask queuedTask = queue.get(galleryTask.getGid()); + if(queuedTask == null){ + refreshDownloadingTask(galleryTask, downloadingDirectory); + GalleryTask existing = queue.putIfAbsent(galleryTask.getGid(), galleryTask); + return existing == null ? galleryTask : existing; + } + if(!queuedTask.is_compressing() && !queuedTask.is_compress_complete()) + refreshDownloadingTask(queuedTask, downloadingDirectory); + return queuedTask; + } + + GalleryTask queuedTask = queue.putIfAbsent(galleryTask.getGid(), galleryTask); + return queuedTask == null ? galleryTask : queuedTask; + } } private void refreshDownloadingTask(GalleryTask galleryTask, File downloadingDirectory){ diff --git a/src/main/java/lion/Service/ReliableTaskService.java b/src/main/java/lion/Service/ReliableTaskService.java index 2df2241..13d8079 100644 --- a/src/main/java/lion/Service/ReliableTaskService.java +++ b/src/main/java/lion/Service/ReliableTaskService.java @@ -67,9 +67,17 @@ public class ReliableTaskService { /** * 启动恢复。 * - *

先读回日志,再把它合并进内存队列;被中断的压缩一律退回「下载完成」,让扫描与压缩 - * 重新评估真实文件,而不是停在一个永远等不到结果的中间态。真实文件状态由后续扫描重算, - * 日志里的进度只作为扫描前的乐观初值。 + *

先读回日志,再按磁盘事实重建每个任务的状态,最后并入内存队列。 + * + *

这里必须用「有效 ZIP / 下载目录」的真实校验来重建,而不是把中断的压缩简单改成 + * 「下载完成」了事:{@code compress()} 已经原子发布 ZIP 并删掉源目录之后、日志尚未落盘时 + * 崩溃,重启后就是一个「日志说压缩中、磁盘上只剩归档、下载目录空无一物」的三方矛盾。 + * 若只降级为下载完成,扫描会因下载目录为空而提前返回、stableop 重投又只回放回执, + * 任务就永久停在等待压缩。按 gid 命中有效归档即判为压缩完成(同时补回 name/path), + * 命中源目录则恢复下载中/下载完成与进度,两者都没有时保留任务、不捏造完成。 + * + *

恢复只重建本地事实,绝不重新提交上游下载;operationId 与待 ACK 报告原样保留, + * 任务的 revision 过滤仍由 {@link #applyAck} 负责。 */ public void restore() throws IOException { journal.load(); @@ -77,11 +85,19 @@ public class ReliableTaskService { for (TaskJournal.JournalTask stored : journal.tasks().values()) { GalleryTask task = stored.toGalleryTask(); if (task.getStatus() == GalleryTask.COMPRESSING) + // 退回到可被扫描重排的状态;此处尚未入队,addToQueue 不会因「正在压缩」短路。 task.setStatus(GalleryTask.DOWNLOAD_COMPLETE); - queue.putIfAbsent(task.getGid(), task); + // 用磁盘事实重建:有效归档 → 压缩完成;源目录 → 下载中/完成(带进度); + // 都没有时保留任务原状。恢复期的命中不应再发「添加时已完成」通知。 + GalleryTask restored = downloadCheckService.addToQueue(task, false); + queue.putIfAbsent(restored.getGid(), restored); if (stored.operationId != null) operationIds.put(stored.gid, stored.operationId); } + // 把重建出的磁盘事实立即落盘:日志停在 COMPRESSING 时它必须被新的压缩完成状态与更高 + // revision 取代。这既让后续报告不再带着过期的压缩中状态上报,也让崩溃前生成的旧快照 + // ACK 在 removeTaskIfRevision 处因版本过期而被拒绝,清不掉刚恢复的新事实。 + journal.syncTasks(List.copyOf(queue.values()), operationIds); log.info("节点重启恢复: 任务={}, 待 ACK 报告={}", queue.size(), journal.pendingReports().size()); } } @@ -138,28 +154,34 @@ public class ReliableTaskService { private byte executeDelete(TaskCommandMessage command) throws IOException { int gid = command.getGid(); - // 压缩中删除会破坏正在进行中的归档写入,先回可重试的「忙」,不写回执。 - if (downloadCheckService.isCompressing(gid)) { - log.info("任务正在压缩,删除请求延后重试: gid={}", gid); - return NodeErrorCode.BUSY_COMPRESSING; - } + // 删除的「查忙 → 删磁盘 → 清队列」必须与扫描的「选任务 → 标压缩中 → 排进压缩队列」 + // 原子互斥。只查一次 isCompressing 会在「已标 COMPRESSING、尚未进压缩队列」的窗口里 + // 判定为不忙:删除把画廊删掉,扫描随后仍把删除前选中的旧对象排进压缩队列,压缩线程 + // 又把归档重新写出来,已删除的画廊被复活并上报。整段持同一监视器后该窗口不复存在。 + synchronized (downloadCheckService) { + // 压缩中删除会破坏正在进行中的归档写入,先回可重试的「忙」,不写回执。 + if (downloadCheckService.isCompressing(gid)) { + log.info("任务正在压缩,删除请求延后重试: gid={}", gid); + return NodeErrorCode.BUSY_COMPRESSING; + } - String directoryName = resolveDirectoryName(gid, command.getName()); - byte result = directoryName == null - ? NodeErrorCode.FILE_NOT_FOUND - : deleteGallery(directoryName); + String directoryName = resolveDirectoryName(gid, command.getName()); + byte result = directoryName == null + ? NodeErrorCode.FILE_NOT_FOUND + : deleteGallery(directoryName); - if (!NodeErrorCode.isTerminal(result)) + if (!NodeErrorCode.isTerminal(result)) + return result; + + downloadCheckService.removeGid(gid); + queue.remove(gid); + operationIds.remove(gid); + journal.removeTask(gid); + journal.putReceipt(TaskJournal.CommandReceipt.of( + command.getOperationId(), TaskCommandMessage.ACTION_DELETE, + gid, directoryName, result)); return result; - - downloadCheckService.removeGid(gid); - queue.remove(gid); - operationIds.remove(gid); - journal.removeTask(gid); - journal.putReceipt(TaskJournal.CommandReceipt.of( - command.getOperationId(), TaskCommandMessage.ACTION_DELETE, - gid, directoryName, result)); - return result; + } } private byte deleteGallery(String name) { diff --git a/src/test/java/lion/Service/DownloadCheckDeleteRaceTest.java b/src/test/java/lion/Service/DownloadCheckDeleteRaceTest.java new file mode 100644 index 0000000..0dac4d4 --- /dev/null +++ b/src/test/java/lion/Service/DownloadCheckDeleteRaceTest.java @@ -0,0 +1,206 @@ +package lion.Service; + +import lion.Domain.GalleryTask; +import lion.Message.Main.TaskCommandMessage; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * 下载扫描与删除的互斥(问题 B)。 + * + *

缺陷:扫描线程在「把任务标成 COMPRESSING」与「排进压缩队列」之间可以被抢占;删除此刻看到 + * active 集合与压缩队列里都没有该 gid,判定为「不忙」并把画廊删掉;扫描随后仍把删除前选中的旧 + * 对象排进压缩队列,压缩线程又把归档写出来,已删除的画廊被复活并上报。 + * + *

仅检查状态位({@code isCompressing})无法修复:窗口正是「状态已改、还没入队」的那段。必须让 + * 「查忙 → 删磁盘 → 清队列」与「选任务 → 标状态 → 入队」落在同一把服务监视器下原子执行。 + * + *

所有交错点用 CountDownLatch / 线程状态确定性控制,不用 sleep 概率等待。 + */ +class DownloadCheckDeleteRaceTest { + + /** + * 端到端交错:扫描已选中任务、正处在「标 COMPRESSING 到入队」的窗口内时,删除不能同时成功。 + * + *

用可控任务在 {@code setStatus(COMPRESSING)} 处暂停——那正是缺陷窗口的位置,且该调用在 + * 缺陷版本与修复版本中同样存在。暂停时删除线程必须阻塞在服务监视器上,直到扫描完成入队; + * 随后删除看到「正在压缩」,回可重试的忙,画廊与归档都不会被破坏。 + * + *

缺陷版本里删除不被监视器阻挡,会在此窗口内直接完成,从而与「扫描已入队压缩」同时成功—— + * 该断言正是对缺陷的判别。 + */ + @Test + void deleteCannotCompleteWhileScanIsBetweenCompressingMarkAndEnqueue(@TempDir Path root) throws Exception { + Path storage = Files.createDirectories(root.resolve("gallery")); + Path downloads = Files.createDirectories(root.resolve("downloads")); + Path source = Files.createDirectories(downloads.resolve("sample [90003]")); + Files.writeString(source.resolve("1.jpg"), "a"); + Files.writeString(source.resolve("galleryinfo.txt"), "meta"); + // 删除目标:gid 命中的画廊目录(压缩归档还未生成)。 + Path gallery = Files.createDirectories(storage.resolve("sample [90003]")); + + Map queue = new ConcurrentHashMap<>(); + DownloadCheckService service = new DownloadCheckService(queue, false); + service.storagePath = storage.toString(); + service.downloadPath = downloads.toString(); + ReliableTaskService reliable = new ReliableTaskService( + root.resolve("journal"), queue, service, storage); + + PausingTask task = new PausingTask(); + task.setGid(90003); + task.setName("sample [90003]"); + task.setPath(source.toString()); + task.setStatus(GalleryTask.DOWNLOAD_COMPLETE); + queue.put(90003, task); + task.monitor = service; + task.arm(); + + Thread scan = new Thread(service::downloadCheck, "scan"); + scan.start(); + assertTrue(task.entered.await(5, TimeUnit.SECONDS), "扫描应到达压缩入队前的标记点"); + assertTrue(task.heldMonitor, "标记点必须位于服务监视器内"); + assertTrue(Files.exists(gallery), "此时画廊目录应仍然存在"); + + byte[] deleteResult = new byte[1]; + Thread delete = new Thread(() -> { + try { + deleteResult[0] = reliable.handleCommand( + deleteCommand("op-race-e2e", 90003, "sample [90003]")); + } catch (Exception e) { + throw new RuntimeException(e); + } + }, "delete"); + delete.start(); + awaitBlocked(delete); + assertTrue(Files.exists(gallery), "扫描持锁期间删除不得移除画廊目录"); + + task.resume.countDown(); + scan.join(5000); + delete.join(5000); + + assertFalse(scan.isAlive()); + assertFalse(delete.isAlive()); + assertEquals(NodeErrorCode.BUSY_COMPRESSING, deleteResult[0], + "扫描已进入压缩调度时删除必须回可重试的忙,两者不能同时成功"); + assertTrue(service.isCompressing(90003), "任务应仍处于压缩调度中"); + + service.compress(); + assertTrue(DownloadCheckService.isValidArchive( + storage.resolve("sample [90003]/sample [90003].zip").toFile()), + "压缩应发布有效归档"); + assertEquals(GalleryTask.COMPRESS_COMPLETE, task.getStatus()); + } + + /** + * 删除已经完成时,随后的扫描不得把已删除任务重新压缩,画廊也不得被复活。 + * + *

覆盖「DELETE 先落地」的时序:入队按 gid 校验队列中的当前对象,避免把删除前遍历到的 + * 旧对象重新排进压缩队列。 + */ + @Test + void scanAfterCompletedDeleteDoesNotCompressOrResurrectGallery(@TempDir Path root) throws Exception { + Fixture fixture = fixture(root, 90004, "sample [90004]"); + Files.createDirectories(root.resolve("gallery/sample [90004]")); + + assertEquals(NodeErrorCode.SUCCESS, + fixture.reliable.handleCommand(deleteCommand("op-e2e-del", 90004, "sample [90004]"))); + assertTrue(fixture.queue.isEmpty(), "删除成功后任务应从队列移除"); + + fixture.service.downloadCheck(); + assertFalse(fixture.service.isCompressing(90004), "已删除任务不得再被排进压缩"); + fixture.service.compress(); + assertFalse(Files.exists(root.resolve("gallery/sample [90004]")), + "已删除的画廊不得被压缩复活"); + } + + // ---- 测试装配 ---- + + private record Fixture(Map queue, DownloadCheckService service, + ReliableTaskService reliable, GalleryTask task, int gid) {} + + private static Fixture fixture(Path root, int gid, String name) throws Exception { + Map queue = new ConcurrentHashMap<>(); + Path storage = Files.createDirectories(root.resolve("gallery")); + Path downloads = Files.createDirectories(root.resolve("downloads")); + Path source = Files.createDirectories(downloads.resolve(name)); + Files.writeString(source.resolve("1.jpg"), "a"); + Files.writeString(source.resolve("galleryinfo.txt"), "meta"); + + DownloadCheckService service = new DownloadCheckService(queue, false); + service.storagePath = storage.toString(); + service.downloadPath = downloads.toString(); + ReliableTaskService reliable = new ReliableTaskService( + root.resolve("journal"), queue, service, storage); + + GalleryTask task = new GalleryTask(); + task.setGid(gid); + task.setName(name); + task.setPath(source.toString()); + task.setStatus(GalleryTask.DOWNLOAD_COMPLETE); + queue.put(gid, task); + return new Fixture(queue, service, reliable, task, gid); + } + + private static TaskCommandMessage deleteCommand(String operationId, int gid, String name) { + TaskCommandMessage command = new TaskCommandMessage(); + command.setOperationId(operationId); + command.setAction(TaskCommandMessage.ACTION_DELETE); + command.setGid(gid); + command.setName(name); + return command; + } + + /** 有界等待线程真正阻塞在监视器上;超时即视为「未按预期互斥」而失败。 */ + private static void awaitBlocked(Thread thread) { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (thread.getState() != Thread.State.BLOCKED) { + if (System.nanoTime() > deadline) + fail("线程未在预期时间内阻塞在服务监视器上,当前状态=" + thread.getState()); + Thread.onSpinWait(); + } + } + + /** + * 可在「标记 COMPRESSING」处暂停的任务。 + * + *

该行存在于缺陷版本与修复版本中,正是「状态已改、尚未入队」的缺陷窗口。 + */ + static final class PausingTask extends GalleryTask { + final CountDownLatch entered = new CountDownLatch(1); + final CountDownLatch resume = new CountDownLatch(1); + volatile boolean armed; + volatile boolean heldMonitor; + + void arm() { + armed = true; + } + + @Override + public void setStatus(byte status) { + if (armed && status == COMPRESSING) { + armed = false; + heldMonitor = Thread.holdsLock(this.monitor); + entered.countDown(); + try { + if (!resume.await(5, TimeUnit.SECONDS)) + throw new IllegalStateException("测试未释放扫描暂停点"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + super.setStatus(status); + } + + /** 服务实例;由测试设置,用于断言暂停点确实位于服务监视器内。 */ + volatile Object monitor; + } +} diff --git a/src/test/java/lion/Service/ReliableTaskServiceTest.java b/src/test/java/lion/Service/ReliableTaskServiceTest.java index abdcbdc..c3d4c15 100644 --- a/src/test/java/lion/Service/ReliableTaskServiceTest.java +++ b/src/test/java/lion/Service/ReliableTaskServiceTest.java @@ -152,6 +152,100 @@ class ReliableTaskServiceTest { assertNotNull(restarted.service.journal().receipt("op-r"), "命令回执必须跨重启保留"); } + /** + * 压缩已原子发布 ZIP 并删除源目录、但 COMPRESS_COMPLETE 尚未同步进 journal 时崩溃。 + * + *

这是问题 A 的核心回归:日志停在 COMPRESSING,磁盘上却已有有效归档、下载目录空空如也。 + * 若重启只把 COMPRESSING 降级成 DOWNLOAD_COMPLETE,扫描会因空下载目录提前返回、重投又只 + * 回放回执,任务永久等不到压缩。恢复必须按 gid 校验磁盘上的有效 ZIP 并重建为压缩完成。 + */ + @Test + void restartRecoversPublishedArchiveAfterCrashBeforeJournalSync(@TempDir Path root) throws Exception { + Harness first = new Harness(root); + first.service.handleCommand(download("op-archive", 80060, "sample [80060]")); + // 模拟状态已推进到压缩中并落盘,随后压缩成功发布归档,但日志未再同步。 + first.queue.get(80060).setStatus(GalleryTask.COMPRESSING); + first.service.refreshJournalFromQueue(); + TaskJournal.ReportSnapshot snapshot = first.service.buildReport().orElseThrow(); + + Path archiveDirectory = Files.createDirectories(first.storagePath.resolve("sample [80060]")); + writeValidZip(archiveDirectory.resolve("sample [80060].zip")); + // 源目录已被 compress() 删除,下载目录为空。 + assertFalse(Files.exists(first.downloadsPath.resolve("sample [80060]"))); + + Harness restarted = new Harness(root, true); + + GalleryTask restored = restarted.queue.get(80060); + assertNotNull(restored, "重启必须保留任务"); + assertEquals(GalleryTask.COMPRESS_COMPLETE, restored.getStatus(), + "磁盘上已有有效归档时必须恢复为压缩完成,而不是停在等待压缩"); + assertEquals("sample [80060]", restored.getName(), "恢复必须重建目录名"); + assertEquals(archiveDirectory.toString(), restored.getPath(), "恢复必须重建归档路径"); + assertTrue(restarted.service.pendingReports().stream() + .anyMatch(report -> report.reportId.equals(snapshot.reportId)), + "恢复必须保留待 ACK 报告"); + assertEquals("op-archive", restarted.service.operationIds().get(80060), + "恢复必须保留 operationId,不得重新提交上游"); + } + + /** 没有有效归档、也没有源目录时:保留任务,但绝不捏造压缩完成。 */ + @Test + void restartWithoutArchiveOrSourceKeepsTaskWithoutFabricatingCompletion(@TempDir Path root) throws Exception { + Harness first = new Harness(root); + first.service.handleCommand(download("op-empty", 80061, "sample [80061]")); + first.queue.get(80061).setStatus(GalleryTask.COMPRESSING); + first.service.refreshJournalFromQueue(); + + Harness restarted = new Harness(root, true); + + GalleryTask restored = restarted.queue.get(80061); + assertNotNull(restored, "无归档且无源时也必须保留任务"); + assertNotEquals(GalleryTask.COMPRESS_COMPLETE, restored.getStatus(), + "没有有效归档时不得捏造压缩完成"); + assertNull(restored.getPath(), "没有真实源目录时不得编造路径"); + } + + /** 压缩中途源目录仍在:恢复为可继续压缩的下载完成态,并带回 path 与进度。 */ + @Test + void restartWithSourcePresentRebuildsDownloadCompleteWithPathAndProgress(@TempDir Path root) throws Exception { + Harness first = new Harness(root); + first.service.handleCommand(download("op-source", 80062, "sample [80062]")); + first.queue.get(80062).setStatus(GalleryTask.COMPRESSING); + first.service.refreshJournalFromQueue(); + + Path source = Files.createDirectories(first.downloadsPath.resolve("sample [80062]")); + Files.writeString(source.resolve("1.jpg"), "a"); + Files.writeString(source.resolve("2.jpg"), "b"); + Files.writeString(source.resolve("galleryinfo.txt"), "meta"); + + Harness restarted = new Harness(root, true); + + GalleryTask restored = restarted.queue.get(80062); + assertEquals(GalleryTask.DOWNLOAD_COMPLETE, restored.getStatus(), + "源目录仍在时应恢复为可重试的下载完成态"); + assertEquals(source.toString(), restored.getPath(), "恢复必须重建源目录路径"); + assertEquals(2, restored.getProceeding(), "恢复必须重建已下载页数"); + } + + /** 同名位置上的 ZIP 已损坏:不得被判为压缩完成,任务必须保留待重试。 */ + @Test + void restartWithCorruptArchiveDoesNotComplete(@TempDir Path root) throws Exception { + Harness first = new Harness(root); + first.service.handleCommand(download("op-corrupt", 80063, "sample [80063]")); + first.queue.get(80063).setStatus(GalleryTask.COMPRESSING); + first.service.refreshJournalFromQueue(); + + Path archiveDirectory = Files.createDirectories(first.storagePath.resolve("sample [80063]")); + Files.writeString(archiveDirectory.resolve("sample [80063].zip"), "partial zip"); + + Harness restarted = new Harness(root, true); + + GalleryTask restored = restarted.queue.get(80063); + assertNotNull(restored, "损坏归档不得让任务消失"); + assertNotEquals(GalleryTask.COMPRESS_COMPLETE, restored.getStatus(), + "损坏或空归档绝不能判为压缩完成"); + } + /** 命令缺 operationId 或 action 非法时回不可重试的无效命令,且不产生回执。 */ @Test void invalidCommandsAreRejectedWithoutReceipt(@TempDir Path root) throws Exception { @@ -167,12 +261,52 @@ class ReliableTaskServiceTest { assertTrue(harness.service.journal().receipts().isEmpty(), "无效命令不得留下回执"); } + /** + * 重启把任务恢复成新事实后,崩溃前生成的旧待 ACK 报告不得清掉它。 + * + *

恢复必须把重建结果落盘({@link ReliableTaskService#restore} 末尾的 syncTasks): + * 它既让日志与磁盘一致(否则后续报告会带着过期的 COMPRESSING 上报),又用更高的 revision + * 取代旧记录,使一份内容巧合匹配的旧 ACK 被 removeTaskIfRevision 的版本过滤拒绝。 + */ + @Test + void staleAckFromBeforeRestartDoesNotClearRecoveredTask(@TempDir Path root) throws Exception { + Harness first = new Harness(root); + first.service.handleCommand(download("op-stale", 80070, "sample [80070]")); + // 崩溃前:日志停在压缩中,且已生成一份覆盖该状态的待 ACK 报告。 + first.queue.get(80070).setStatus(GalleryTask.COMPRESSING); + first.service.refreshJournalFromQueue(); + TaskJournal.ReportSnapshot staleSnapshot = first.service.buildReport().orElseThrow(); + + Path archiveDirectory = Files.createDirectories(first.storagePath.resolve("sample [80070]")); + writeValidZip(archiveDirectory.resolve("sample [80070].zip")); + + Harness restarted = new Harness(root, true); + GalleryTask recovered = restarted.queue.get(80070); + assertEquals(GalleryTask.COMPRESS_COMPLETE, recovered.getStatus(), "重启应按磁盘事实恢复为压缩完成"); + // 恢复结果必须已经落盘:从磁盘重新加载日志仍是新状态。 + TaskJournal reloaded = new TaskJournal(root.resolve("task-journal")); + reloaded.load(); + assertEquals(GalleryTask.COMPRESS_COMPLETE, reloaded.task(80070).status, + "恢复后的新事实必须同步进日志"); + assertTrue(reloaded.task(80070).revision + > staleSnapshot.tasks.stream().filter(t -> t.gid == 80070).findFirst().orElseThrow().revision, + "恢复必须推进 revision,使旧快照的版本过滤失效"); + + restarted.service.applyAck(staleSnapshot.reportId, NodeErrorCode.SUCCESS); + + assertTrue(restarted.queue.containsKey(80070), + "崩溃前的旧 ACK 不得清掉已恢复的任务"); + assertTrue(restarted.service.journal().tasks().containsKey(80070), + "日志里也必须保留已恢复的任务"); + } + /** 一个最小离线装配:显式目录、无调度器、无网络。 */ static class Harness { final Map queue = new ConcurrentHashMap<>(); final DownloadCheckService downloadCheck; final ReliableTaskService service; final Path storagePath; + final Path downloadsPath; Harness(Path root) throws Exception { this(root, false); @@ -180,17 +314,28 @@ class ReliableTaskServiceTest { Harness(Path root, boolean restore) throws Exception { storagePath = Files.createDirectories(root.resolve("gallery")); - Path downloads = Files.createDirectories(root.resolve("downloads")); + downloadsPath = Files.createDirectories(root.resolve("downloads")); Path journalDir = root.resolve("task-journal"); downloadCheck = new DownloadCheckService(queue, false); downloadCheck.storagePath = storagePath.toString(); - downloadCheck.downloadPath = downloads.toString(); + downloadCheck.downloadPath = downloadsPath.toString(); service = new ReliableTaskService(journalDir, queue, downloadCheck, storagePath); if (restore) service.restore(); } } + /** 写一个真实的小 ZIP(含 CRC 校验,可被 isValidArchive 接受)。 */ + static void writeValidZip(Path zip) throws Exception { + Files.createDirectories(zip.getParent()); + try (java.util.zip.ZipOutputStream out = + new java.util.zip.ZipOutputStream(Files.newOutputStream(zip))) { + out.putNextEntry(new java.util.zip.ZipEntry("1.jpg")); + out.write("image bytes".getBytes(java.nio.charset.StandardCharsets.UTF_8)); + out.closeEntry(); + } + } + private static TaskCommandMessage download(String operationId, int gid, String name) { TaskCommandMessage command = new TaskCommandMessage(); command.setOperationId(operationId);